orch/store/
dependencies.rs

1//! Issue dependency rows and the claim gate that keeps a waiting task from starting.
2//!
3//! A dependency is done when its blocker snapshot says the blocking pull request
4//! was merged into the default branch, or the issue has no pull request and was
5//! closed as completed. A local `done` row is not enough.
6//! `not_planned` and `duplicate` are not done. `new`, `routed`, `waiting`,
7//! `blocked`, and `needs_review` move to `waiting` while a dependency is open.
8//! `in_progress` and `in_review` stay where they are.
9//! A `blocked-external` note in the issue body is an unmet blocker until a later
10//! import removes it
11
12use super::{TaskStatus, TaskStore};
13use crate::cluster::Command;
14use anyhow::Context;
15use sqlx::Row;
16use sqlx::SqliteConnection;
17use std::collections::HashMap;
18
19/// Returned by routing and dispatch when a task still has an open dependency.
20///
21/// Callers log this and leave the task alone. It does not count as a routing failure
22#[derive(Debug, thiserror::Error)]
23#[error("waiting on {}", .blockers.join(", "))]
24pub struct WaitingOnDependency {
25    pub blockers: Vec<String>,
26}
27
28#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
29pub enum DepSource {
30    Github,
31    Body,
32    Cli,
33}
34
35impl DepSource {
36    pub fn as_str(self) -> &'static str {
37        match self {
38            Self::Github => "github",
39            Self::Body => "body",
40            Self::Cli => "cli",
41        }
42    }
43
44    pub fn parse(value: &str) -> Option<Self> {
45        match value {
46            "github" => Some(Self::Github),
47            "body" => Some(Self::Body),
48            "cli" => Some(Self::Cli),
49            _ => None,
50        }
51    }
52}
53
54#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
55pub struct NewDependency {
56    pub repo: String,
57    pub number: i64,
58    pub source: DepSource,
59}
60
61#[derive(Debug, Clone)]
62pub struct TaskDependency {
63    pub task_id: i64,
64    pub depends_on_task_id: Option<i64>,
65    pub depends_on_repo: String,
66    pub depends_on_number: i64,
67    pub source: String,
68    pub blocker_status: Option<String>,
69}
70
71#[derive(Debug, Clone)]
72pub struct SchedulableTask {
73    pub id: i64,
74    pub repo: String,
75    pub status: String,
76    pub external_id: Option<String>,
77    pub body: String,
78    pub title: String,
79}
80
81#[derive(Debug, Clone)]
82pub struct DepEdge {
83    pub task_id: i64,
84    pub task_repo: String,
85    pub task_external_id: Option<String>,
86    pub task_title: String,
87    pub depends_on_repo: String,
88    pub depends_on_number: i64,
89}
90
91#[derive(Debug, Clone)]
92struct BlockerView {
93    label: String,
94    satisfied: bool,
95}
96
97/// Pull request fields the engine observed on GitHub.
98///
99/// The store does not talk to GitHub. The engine fills this and the snapshot
100/// decides whether the blocker is done
101#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
102pub struct ObservedPullRequest {
103    pub merged: bool,
104    pub base_is_default: bool,
105    pub approved: bool,
106    pub closed_unmerged: bool,
107}
108
109/// What orch last read about one blocking issue.
110///
111/// Missing JSON fails closed: the dependency stays open
112#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
113pub struct BlockerSnapshot {
114    pub issue_state: String,
115    pub state_reason: Option<String>,
116    pub has_pr: bool,
117    pub pr_merged: bool,
118    pub pr_merged_into_default: bool,
119    pub pr_approved: bool,
120    pub pr_closed: bool,
121}
122
123impl BlockerSnapshot {
124    pub fn observe(
125        issue_state: &str,
126        state_reason: Option<&str>,
127        pr: Option<&ObservedPullRequest>,
128    ) -> Self {
129        let normalized = normalize_close_reason(issue_state, state_reason);
130        let (has_pr, merged, into_default, approved, pr_closed) = match pr {
131            Some(pr) => (
132                true,
133                pr.merged,
134                pr.base_is_default,
135                pr.approved,
136                pr.closed_unmerged,
137            ),
138            None => (false, false, false, false, false),
139        };
140        Self {
141            issue_state: if issue_state.eq_ignore_ascii_case("closed") {
142                "closed".to_string()
143            } else {
144                "open".to_string()
145            },
146            state_reason: Some(normalized.to_string()),
147            has_pr,
148            pr_merged: merged,
149            pr_merged_into_default: into_default,
150            pr_approved: approved,
151            pr_closed,
152        }
153    }
154
155    /// Record a new issue state without clearing pull request fields.
156    ///
157    /// A later issue sync must not wipe a snapshot that already knows the PR
158    pub fn note_issue_state(&mut self, issue_state: &str, state_reason: Option<&str>) {
159        let normalized = normalize_close_reason(issue_state, state_reason);
160        self.issue_state = if issue_state.eq_ignore_ascii_case("closed") {
161            "closed".to_string()
162        } else {
163            "open".to_string()
164        };
165        self.state_reason = Some(normalized.to_string());
166    }
167
168    pub fn is_satisfied(&self) -> bool {
169        let reason = self.state_reason.as_deref().unwrap_or("");
170        if reason == "not_planned" || reason == "duplicate" {
171            return false;
172        }
173        if self.has_pr {
174            // A merge into the default branch is the decision, even without an approving review
175            return self.pr_merged && self.pr_merged_into_default;
176        }
177        self.issue_state == "closed" && (reason.is_empty() || reason == "completed")
178    }
179}
180
181/// Map a GitHub issue `state` and `state_reason` onto the values stored in KV.
182///
183/// A classic close with no reason counts as completed. An open issue counts as open
184pub fn normalize_close_reason(state: &str, state_reason: Option<&str>) -> &'static str {
185    if !state.eq_ignore_ascii_case("closed") {
186        return "open";
187    }
188    match state_reason
189        .unwrap_or("")
190        .trim()
191        .to_ascii_lowercase()
192        .as_str()
193    {
194        "not_planned" => "not_planned",
195        "duplicate" => "duplicate",
196        "reopened" => "open",
197        _ => "completed",
198    }
199}
200
201pub fn blocker_snapshot_key(repo: &str, number: i64) -> String {
202    format!("dep_blocker:{repo}#{number}")
203}
204
205pub fn worktree_refresh_key(task_id: i64) -> String {
206    format!("dep_worktree_refresh:{task_id}")
207}
208
209pub fn dep_label(task_repo: &str, dep_repo: &str, number: i64) -> String {
210    if task_repo == dep_repo {
211        format!("#{number}")
212    } else {
213        format!("{dep_repo}#{number}")
214    }
215}
216
217/// A missing snapshot is not done. Local task status is not an input
218pub fn blocker_is_satisfied(snapshot: Option<&BlockerSnapshot>) -> bool {
219    snapshot.is_some_and(BlockerSnapshot::is_satisfied)
220}
221
222/// `not_planned` and a still-open `blocked` task each get one notice.
223///
224/// `duplicate` does not. No snapshot means no notice
225pub fn blocker_failure_kind(
226    task_status: Option<&str>,
227    snapshot: Option<&BlockerSnapshot>,
228) -> Option<&'static str> {
229    let snap = snapshot?;
230    if snap.state_reason.as_deref() == Some("not_planned") {
231        return Some("not_planned");
232    }
233    if task_status == Some("blocked") && !snap.is_satisfied() {
234        return Some("blocked");
235    }
236    None
237}
238
239impl TaskStore {
240    /// Rename a repo slug on every stored row in one transaction.
241    ///
242    /// Refuses when the new slug already has tasks. Returns the number of moved tasks
243    pub async fn move_repo(&self, old: &str, new: &str) -> anyhow::Result<u64> {
244        let cmd = Command::MoveRepo {
245            old: old.to_string(),
246            new: new.to_string(),
247        };
248        Ok(u64::try_from(self.write_int(cmd).await?)?)
249    }
250
251    /// Replace one task's dependency rows in a single transaction.
252    pub async fn replace_dependencies(
253        &self,
254        task_id: i64,
255        rows: &[NewDependency],
256    ) -> anyhow::Result<()> {
257        let cmd = Command::DepReplace {
258            task_id,
259            rows: rows.to_vec(),
260        };
261        self.write(cmd).await.map(|_| ())
262    }
263
264    /// Point dependency rows at a task orch now knows.
265    pub async fn link_known_task(
266        &self,
267        task_id: i64,
268        repo: &str,
269        number: i64,
270    ) -> anyhow::Result<()> {
271        let cmd = Command::DepLinkKnown {
272            task_id,
273            repo: repo.to_string(),
274            number,
275        };
276        self.write(cmd).await.map(|_| ())
277    }
278
279    pub async fn list_dependencies(&self, task_id: i64) -> anyhow::Result<Vec<TaskDependency>> {
280        let rows = sqlx::query(
281            "SELECT d.task_id, d.depends_on_task_id, d.depends_on_repo, d.depends_on_number, d.source,
282                    t.status AS blocker_status
283             FROM task_dependencies d
284             LEFT JOIN tasks t
285               ON t.repo = d.depends_on_repo
286              AND t.external_id = CAST(d.depends_on_number AS TEXT)
287             WHERE d.task_id = ?
288             ORDER BY d.depends_on_repo, d.depends_on_number",
289        )
290        .bind(task_id)
291        .fetch_all(&self.pool)
292        .await?;
293        let mut out = Vec::with_capacity(rows.len());
294        for row in rows {
295            out.push(TaskDependency {
296                task_id: row.try_get("task_id")?,
297                depends_on_task_id: row.try_get("depends_on_task_id")?,
298                depends_on_repo: row.try_get("depends_on_repo")?,
299                depends_on_number: row.try_get("depends_on_number")?,
300                source: row.try_get("source")?,
301                blocker_status: row.try_get("blocker_status")?,
302            });
303        }
304        Ok(out)
305    }
306
307    pub async fn list_dependents(
308        &self,
309        repo: &str,
310        number: i64,
311    ) -> anyhow::Result<Vec<TaskDependency>> {
312        let rows = sqlx::query(
313            "SELECT d.task_id, d.depends_on_task_id, d.depends_on_repo, d.depends_on_number, d.source,
314                    t.status AS blocker_status
315             FROM task_dependencies d
316             JOIN tasks t ON t.id = d.task_id
317             WHERE d.depends_on_repo = ? AND d.depends_on_number = ?
318             ORDER BY d.task_id",
319        )
320        .bind(repo)
321        .bind(number)
322        .fetch_all(&self.pool)
323        .await?;
324        let mut out = Vec::with_capacity(rows.len());
325        for row in rows {
326            out.push(TaskDependency {
327                task_id: row.try_get("task_id")?,
328                depends_on_task_id: row.try_get("depends_on_task_id")?,
329                depends_on_repo: row.try_get("depends_on_repo")?,
330                depends_on_number: row.try_get("depends_on_number")?,
331                source: row.try_get("source")?,
332                blocker_status: row.try_get("blocker_status")?,
333            });
334        }
335        Ok(out)
336    }
337
338    pub async fn list_schedulable(&self) -> anyhow::Result<Vec<SchedulableTask>> {
339        let rows = sqlx::query(
340            "SELECT id, repo, status, external_id, body, title
341             FROM tasks
342             WHERE status IN ('new', 'routed', 'waiting')
343             ORDER BY id",
344        )
345        .fetch_all(&self.pool)
346        .await?;
347        let mut out = Vec::with_capacity(rows.len());
348        for row in rows {
349            out.push(SchedulableTask {
350                id: row.try_get("id")?,
351                repo: row.try_get("repo")?,
352                status: row.try_get("status")?,
353                external_id: row.try_get("external_id")?,
354                body: row.try_get("body")?,
355                title: row.try_get("title")?,
356            });
357        }
358        Ok(out)
359    }
360
361    pub async fn list_dependency_edges(&self) -> anyhow::Result<Vec<DepEdge>> {
362        let rows = sqlx::query(
363            "SELECT d.task_id, t.repo AS task_repo, t.external_id AS task_external_id, t.title AS task_title,
364                    d.depends_on_repo, d.depends_on_number
365             FROM task_dependencies d
366             JOIN tasks t ON t.id = d.task_id",
367        )
368        .fetch_all(&self.pool)
369        .await?;
370        let mut out = Vec::with_capacity(rows.len());
371        for row in rows {
372            out.push(DepEdge {
373                task_id: row.try_get("task_id")?,
374                task_repo: row.try_get("task_repo")?,
375                task_external_id: row.try_get("task_external_id")?,
376                task_title: row.try_get("task_title")?,
377                depends_on_repo: row.try_get("depends_on_repo")?,
378                depends_on_number: row.try_get("depends_on_number")?,
379            });
380        }
381        Ok(out)
382    }
383
384    /// Labels of unmet blockers, keyed by the waiting task id.
385    pub async fn unmet_labels_by_task(&self) -> anyhow::Result<HashMap<i64, Vec<String>>> {
386        let mut conn = self.pool.acquire().await?;
387        let ids: Vec<i64> = sqlx::query("SELECT DISTINCT task_id FROM task_dependencies")
388            .fetch_all(&mut *conn)
389            .await?
390            .into_iter()
391            .map(|row| row.try_get("task_id"))
392            .collect::<Result<Vec<_>, _>>()?;
393        let mut out = HashMap::new();
394        for task_id in ids {
395            let repo: String = sqlx::query("SELECT repo FROM tasks WHERE id = ?")
396                .bind(task_id)
397                .fetch_optional(&mut *conn)
398                .await?
399                .and_then(|row| row.try_get("repo").ok())
400                .unwrap_or_default();
401            let views = load_blockers(&mut conn, task_id, &repo).await?;
402            let labels: Vec<String> = views
403                .into_iter()
404                .filter(|view| !view.satisfied)
405                .map(|view| view.label)
406                .collect();
407            if !labels.is_empty() {
408                out.insert(task_id, labels);
409            }
410        }
411        Ok(out)
412    }
413
414    pub async fn unmet_blocker_labels(&self, task_id: i64) -> anyhow::Result<Vec<String>> {
415        let mut conn = self.pool.acquire().await?;
416        let repo: String = sqlx::query("SELECT repo FROM tasks WHERE id = ?")
417            .bind(task_id)
418            .fetch_optional(&mut *conn)
419            .await?
420            .and_then(|row| row.try_get("repo").ok())
421            .unwrap_or_default();
422        let views = load_blockers(&mut conn, task_id, &repo).await?;
423        Ok(views
424            .into_iter()
425            .filter(|view| !view.satisfied)
426            .map(|view| view.label)
427            .collect())
428    }
429
430    pub async fn save_blocker_snapshot(
431        &self,
432        repo: &str,
433        number: i64,
434        snap: &BlockerSnapshot,
435    ) -> anyhow::Result<()> {
436        let raw = serde_json::to_string(snap).context("encode blocker snapshot")?;
437        self.kv_set(&blocker_snapshot_key(repo, number), &raw).await
438    }
439
440    pub async fn blocker_snapshot(
441        &self,
442        repo: &str,
443        number: i64,
444    ) -> anyhow::Result<Option<BlockerSnapshot>> {
445        let Some(raw) = self.kv_get(&blocker_snapshot_key(repo, number)).await? else {
446            return Ok(None);
447        };
448        Ok(serde_json::from_str(&raw).ok())
449    }
450
451    /// Merge an issue state into the stored snapshot. Pull request fields stay.
452    pub async fn note_blocker_issue_state(
453        &self,
454        repo: &str,
455        number: i64,
456        state: &str,
457        reason: Option<&str>,
458    ) -> anyhow::Result<()> {
459        let mut snap = match self.blocker_snapshot(repo, number).await? {
460            Some(existing) => existing,
461            None => BlockerSnapshot::observe(state, reason, None),
462        };
463        snap.note_issue_state(state, reason);
464        self.save_blocker_snapshot(repo, number, &snap).await
465    }
466
467    /// Hold the task when a dependency is open.
468    ///
469    /// `None` means the task may be claimed. `Some` means it is waiting, and the
470    /// vec is the blocking issue labels. The check and the status write share one
471    /// immediate transaction, so two ticks cannot both route the same waiting task.
472    /// `in_progress` and `in_review` are not rewritten
473    pub async fn block_claim_if_waiting(
474        &self,
475        task_id: i64,
476    ) -> anyhow::Result<Option<Vec<String>>> {
477        match self.write(Command::DepBlockClaim { task_id }).await? {
478            crate::cluster::Reply::Unmet(unmet) => Ok(unmet),
479            other => anyhow::bail!("block_claim_if_waiting got {other:?}"),
480        }
481    }
482
483    /// Where `unblock` should leave a task. Waiting when any dependency is still open
484    pub async fn status_after_unblock(&self, task_id: i64) -> anyhow::Result<TaskStatus> {
485        match self.block_claim_if_waiting(task_id).await? {
486            Some(_) => Ok(TaskStatus::Waiting),
487            None => Ok(TaskStatus::New),
488        }
489    }
490}
491
492/// Plan-import note that a blocker outside the milestone has no GitHub issue yet
493pub fn blocked_external_from_body(body: &str) -> Vec<String> {
494    let mut ids = Vec::new();
495    let mut push = |token: &str| {
496        let token = token.trim();
497        if token.is_empty() {
498            return;
499        }
500        if !ids.iter().any(|have| have == token) {
501            ids.push(token.to_string());
502        }
503    };
504    for line in body.lines() {
505        let trimmed = line.trim();
506        if let Some(rest) = trimmed.strip_prefix("<!--") {
507            let rest = rest.trim();
508            let Some(rest) = rest.strip_prefix("orch-blocked-external:") else {
509                continue;
510            };
511            let rest = rest.trim().trim_end_matches("-->").trim();
512            for token in rest.split(',') {
513                push(token);
514            }
515            continue;
516        }
517        let lower = trimmed.to_ascii_lowercase();
518        if !lower.starts_with("blocked-external") {
519            continue;
520        }
521        let rest = trimmed["blocked-external".len()..]
522            .trim_start()
523            .trim_start_matches(':')
524            .trim();
525        let rest = rest.strip_suffix("(no issue yet)").unwrap_or(rest).trim();
526        for token in rest.split(',') {
527            push(token);
528        }
529    }
530    ids
531}
532
533async fn load_blockers(
534    conn: &mut SqliteConnection,
535    task_id: i64,
536    task_repo: &str,
537) -> anyhow::Result<Vec<BlockerView>> {
538    let rows = sqlx::query(
539        "SELECT depends_on_repo, depends_on_number
540         FROM task_dependencies
541         WHERE task_id = ?
542         ORDER BY depends_on_repo, depends_on_number",
543    )
544    .bind(task_id)
545    .fetch_all(&mut *conn)
546    .await?;
547    let mut views = Vec::with_capacity(rows.len());
548    for row in rows {
549        let dep_repo: String = row.try_get("depends_on_repo")?;
550        let number: i64 = row.try_get("depends_on_number")?;
551        let raw: Option<String> = sqlx::query("SELECT value FROM kv WHERE key = ?")
552            .bind(blocker_snapshot_key(&dep_repo, number))
553            .fetch_optional(&mut *conn)
554            .await?
555            .and_then(|found| found.try_get("value").ok());
556        let snap = raw.and_then(|text| serde_json::from_str::<BlockerSnapshot>(&text).ok());
557        views.push(BlockerView {
558            label: dep_label(task_repo, &dep_repo, number),
559            satisfied: snap.as_ref().is_some_and(BlockerSnapshot::is_satisfied),
560        });
561    }
562    Ok(views)
563}
564
565#[cfg(test)]
566mod tests {
567    use super::*;
568    use crate::store::{NewTask, TaskStore};
569
570    async fn task(store: &TaskStore, repo: &str, number: &str, title: &str) -> i64 {
571        task_with_body(store, repo, number, title, "").await
572    }
573
574    async fn task_with_body(
575        store: &TaskStore,
576        repo: &str,
577        number: &str,
578        title: &str,
579        body: &str,
580    ) -> i64 {
581        store
582            .create(&NewTask {
583                external_id: Some(number.to_string()),
584                repo: repo.to_string(),
585                origin: "github".to_string(),
586                title: title.to_string(),
587                body: body.to_string(),
588                source: "manual".to_string(),
589                source_id: String::new(),
590                author: String::new(),
591                url: String::new(),
592                labels: vec![],
593                parent_id: None,
594            })
595            .await
596            .unwrap()
597    }
598
599    fn approved_merge(into_default: bool) -> BlockerSnapshot {
600        BlockerSnapshot::observe(
601            "closed",
602            Some("completed"),
603            Some(&ObservedPullRequest {
604                merged: true,
605                base_is_default: into_default,
606                approved: true,
607                closed_unmerged: false,
608            }),
609        )
610    }
611
612    #[tokio::test]
613    async fn unmet_dependency_demotes_new_and_promotes_when_the_pr_is_done() {
614        let store = TaskStore::open_memory().await.unwrap();
615        let a = task(&store, "owner/repo", "1", "A").await;
616        let b = task(&store, "owner/repo", "2", "B").await;
617        store
618            .set_fields(
619                b,
620                &[
621                    ("branch", serde_json::json!("old-branch")),
622                    ("worktree", serde_json::json!("/tmp/old")),
623                ],
624            )
625            .await
626            .unwrap();
627        store
628            .replace_dependencies(
629                b,
630                &[NewDependency {
631                    repo: "owner/repo".to_string(),
632                    number: 1,
633                    source: DepSource::Github,
634                }],
635            )
636            .await
637            .unwrap();
638
639        let held = store.block_claim_if_waiting(b).await.unwrap();
640        assert_eq!(held, Some(vec!["#1".to_string()]));
641        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
642
643        store.update_status(a, TaskStatus::Done).await.unwrap();
644        assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
645        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
646
647        store
648            .save_blocker_snapshot("owner/repo", 1, &approved_merge(true))
649            .await
650            .unwrap();
651        let released = store.block_claim_if_waiting(b).await.unwrap();
652        assert_eq!(released, None);
653        let task = store.get(b).await.unwrap();
654        assert_eq!(task.status, TaskStatus::New);
655        assert!(task.branch.is_empty());
656        assert!(task.worktree.is_empty());
657        assert_eq!(
658            store
659                .kv_get(&worktree_refresh_key(b))
660                .await
661                .unwrap()
662                .as_deref(),
663            Some("1")
664        );
665    }
666
667    #[tokio::test]
668    async fn not_planned_beats_a_merged_pr_and_a_completed_issue_with_no_pr_is_done() {
669        let store = TaskStore::open_memory().await.unwrap();
670        let a = task(&store, "owner/repo", "1", "A").await;
671        let b = task(&store, "owner/repo", "2", "B").await;
672        store.update_status(a, TaskStatus::Done).await.unwrap();
673        let mut not_planned = approved_merge(true);
674        not_planned.state_reason = Some("not_planned".to_string());
675        store
676            .save_blocker_snapshot("owner/repo", 1, &not_planned)
677            .await
678            .unwrap();
679        store
680            .replace_dependencies(
681                b,
682                &[NewDependency {
683                    repo: "owner/repo".to_string(),
684                    number: 1,
685                    source: DepSource::Github,
686                }],
687            )
688            .await
689            .unwrap();
690        assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
691
692        let c = task(&store, "owner/repo", "3", "C").await;
693        store
694            .save_blocker_snapshot(
695                "other/repo",
696                9,
697                &BlockerSnapshot::observe("closed", Some("completed"), None),
698            )
699            .await
700            .unwrap();
701        store
702            .replace_dependencies(
703                c,
704                &[NewDependency {
705                    repo: "other/repo".to_string(),
706                    number: 9,
707                    source: DepSource::Body,
708                }],
709            )
710            .await
711            .unwrap();
712        assert_eq!(store.block_claim_if_waiting(c).await.unwrap(), None);
713    }
714
715    #[test]
716    fn pull_request_completion_cases() {
717        let open = BlockerSnapshot::observe("open", None, None);
718        assert!(!open.is_satisfied());
719
720        let pr_open = BlockerSnapshot::observe(
721            "open",
722            None,
723            Some(&ObservedPullRequest {
724                merged: false,
725                base_is_default: true,
726                approved: false,
727                closed_unmerged: false,
728            }),
729        );
730        assert!(!pr_open.is_satisfied());
731
732        let approved_open = BlockerSnapshot::observe(
733            "open",
734            None,
735            Some(&ObservedPullRequest {
736                merged: false,
737                base_is_default: true,
738                approved: true,
739                closed_unmerged: false,
740            }),
741        );
742        assert!(!approved_open.is_satisfied());
743
744        assert!(approved_merge(true).is_satisfied());
745        assert!(!approved_merge(false).is_satisfied());
746
747        let closed_unmerged = BlockerSnapshot::observe(
748            "closed",
749            Some("completed"),
750            Some(&ObservedPullRequest {
751                merged: false,
752                base_is_default: true,
753                approved: true,
754                closed_unmerged: true,
755            }),
756        );
757        assert!(!closed_unmerged.is_satisfied());
758
759        let merged_unapproved = BlockerSnapshot::observe(
760            "closed",
761            Some("completed"),
762            Some(&ObservedPullRequest {
763                merged: true,
764                base_is_default: true,
765                approved: false,
766                closed_unmerged: false,
767            }),
768        );
769        assert!(merged_unapproved.is_satisfied());
770
771        assert!(BlockerSnapshot::observe("closed", None, None).is_satisfied());
772        assert!(BlockerSnapshot::observe("closed", Some("completed"), None).is_satisfied());
773        assert!(!BlockerSnapshot::observe("closed", Some("not_planned"), None).is_satisfied());
774        assert!(!BlockerSnapshot::observe("closed", Some("duplicate"), None).is_satisfied());
775    }
776
777    #[tokio::test]
778    async fn blocked_and_duplicate_stay_unmet_and_in_progress_is_not_rewritten() {
779        let store = TaskStore::open_memory().await.unwrap();
780        let a = task(&store, "owner/repo", "1", "A").await;
781        let b = task(&store, "owner/repo", "2", "B").await;
782        store.update_status(a, TaskStatus::Blocked).await.unwrap();
783        store
784            .replace_dependencies(
785                b,
786                &[NewDependency {
787                    repo: "owner/repo".to_string(),
788                    number: 1,
789                    source: DepSource::Cli,
790                }],
791            )
792            .await
793            .unwrap();
794        store
795            .update_status(b, TaskStatus::InProgress)
796            .await
797            .unwrap();
798        assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
799        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::InProgress);
800
801        store
802            .save_blocker_snapshot(
803                "owner/repo",
804                1,
805                &BlockerSnapshot::observe("closed", Some("duplicate"), None),
806            )
807            .await
808            .unwrap();
809        store.update_status(a, TaskStatus::Done).await.unwrap();
810        assert!(store.block_claim_if_waiting(b).await.unwrap().is_some());
811        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::InProgress);
812    }
813
814    #[tokio::test]
815    async fn concurrent_claims_both_report_waiting() {
816        let store = std::sync::Arc::new(TaskStore::open_memory().await.unwrap());
817        let b = task(&store, "owner/repo", "2", "B").await;
818        store
819            .replace_dependencies(
820                b,
821                &[NewDependency {
822                    repo: "owner/repo".to_string(),
823                    number: 1,
824                    source: DepSource::Github,
825                }],
826            )
827            .await
828            .unwrap();
829        let left = store.clone();
830        let right = store.clone();
831        let (a, c) = tokio::join!(
832            left.block_claim_if_waiting(b),
833            right.block_claim_if_waiting(b)
834        );
835        assert!(a.unwrap().is_some());
836        assert!(c.unwrap().is_some());
837        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
838    }
839
840    #[tokio::test]
841    async fn unblock_selection_skips_waiting_and_blocked_with_deps_stays_waiting() {
842        let store = TaskStore::open_memory().await.unwrap();
843        let a = task(&store, "owner/repo", "1", "A").await;
844        let b = task(&store, "owner/repo", "2", "B").await;
845        store.update_status(a, TaskStatus::Blocked).await.unwrap();
846        store
847            .replace_dependencies(
848                b,
849                &[NewDependency {
850                    repo: "owner/repo".to_string(),
851                    number: 1,
852                    source: DepSource::Github,
853                }],
854            )
855            .await
856            .unwrap();
857        store.block_claim_if_waiting(b).await.unwrap();
858        let blocked = store
859            .list_by_status("owner/repo", TaskStatus::Blocked)
860            .await
861            .unwrap();
862        assert!(blocked.iter().all(|task| task.id != b));
863        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
864        assert_eq!(
865            store.status_after_unblock(a).await.unwrap(),
866            TaskStatus::New
867        );
868        store.update_status(a, TaskStatus::Blocked).await.unwrap();
869        store
870            .replace_dependencies(
871                a,
872                &[NewDependency {
873                    repo: "owner/repo".to_string(),
874                    number: 9,
875                    source: DepSource::Cli,
876                }],
877            )
878            .await
879            .unwrap();
880        assert_eq!(
881            store.status_after_unblock(a).await.unwrap(),
882            TaskStatus::Waiting
883        );
884    }
885
886    #[tokio::test]
887    async fn entering_waiting_clears_a_stored_branch() {
888        let store = TaskStore::open_memory().await.unwrap();
889        let b = task(&store, "owner/repo", "2", "B").await;
890        store
891            .set_fields(
892                b,
893                &[
894                    ("branch", serde_json::json!("old-branch")),
895                    ("worktree", serde_json::json!("/tmp/old")),
896                ],
897            )
898            .await
899            .unwrap();
900        store
901            .replace_dependencies(
902                b,
903                &[NewDependency {
904                    repo: "owner/repo".to_string(),
905                    number: 1,
906                    source: DepSource::Github,
907                }],
908            )
909            .await
910            .unwrap();
911        store.block_claim_if_waiting(b).await.unwrap();
912        let task = store.get(b).await.unwrap();
913        assert_eq!(task.status, TaskStatus::Waiting);
914        assert!(task.branch.is_empty());
915        assert!(task.worktree.is_empty());
916    }
917
918    #[test]
919    fn blocked_external_from_body_reads_the_note_and_the_marker() {
920        let body = concat!(
921            "blocked-external: M1-05, M1-01 (no issue yet)\n\n",
922            "<!-- orch-blocked-external: M1-05, M1-01 -->\n",
923            "<!-- orch-plan: M6-01 -->\n",
924        );
925        assert_eq!(
926            blocked_external_from_body(body),
927            vec!["M1-05".to_string(), "M1-01".to_string()]
928        );
929        assert!(blocked_external_from_body("<!-- orch-plan: M6-01 -->").is_empty());
930    }
931
932    #[tokio::test]
933    async fn blocked_external_note_holds_the_task_and_clears_when_removed() {
934        let store = TaskStore::open_memory().await.unwrap();
935        let body = concat!(
936            "blocked-external: M3-01 (no issue yet)\n\n",
937            "<!-- orch-blocked-external: M3-01 -->\n",
938        );
939        let b = task_with_body(&store, "owner/repo", "4", "M4-01 API", body).await;
940        let held = store.block_claim_if_waiting(b).await.unwrap();
941        assert_eq!(held, Some(vec!["M3-01".to_string()]));
942        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::Waiting);
943
944        store
945            .upsert_external(&crate::store::UpsertExternal {
946                repo: "owner/repo",
947                ext_id: "4",
948                title: "M4-01 API",
949                body: "<!-- orch-plan: M4-01 -->\n",
950                author: "",
951                url: "",
952                labels: &[],
953                origin: "github",
954            })
955            .await
956            .unwrap();
957        let released = store.block_claim_if_waiting(b).await.unwrap();
958        assert_eq!(released, None);
959        assert_eq!(store.get(b).await.unwrap().status, TaskStatus::New);
960    }
961
962    #[tokio::test]
963    async fn migration_028_applies_over_existing_tasks() {
964        let dir = std::path::Path::new(env!("CARGO_MANIFEST_DIR")).join("migrations");
965        let options = sqlx::sqlite::SqliteConnectOptions::new()
966            .filename(":memory:")
967            .pragma("foreign_keys", "ON");
968        let pool = sqlx::sqlite::SqlitePoolOptions::new()
969            .max_connections(1)
970            .connect_with(options)
971            .await
972            .unwrap();
973        let mut files: Vec<_> = std::fs::read_dir(&dir)
974            .unwrap()
975            .map(|entry| entry.unwrap().path())
976            .filter(|path| path.extension().and_then(|ext| ext.to_str()) == Some("sql"))
977            .collect();
978        files.sort();
979        for path in &files {
980            let name = path.file_name().unwrap().to_string_lossy();
981            if name.starts_with("028") {
982                continue;
983            }
984            let sql = std::fs::read_to_string(path).unwrap();
985            sqlx::raw_sql(&sql)
986                .execute(&pool)
987                .await
988                .unwrap_or_else(|err| panic!("migration {name} failed: {err}"));
989        }
990        sqlx::query("INSERT INTO tasks (repo, title) VALUES ('owner/repo', 'existing')")
991            .execute(&pool)
992            .await
993            .unwrap();
994        let sql = std::fs::read_to_string(dir.join("028_task_dependencies.sql")).unwrap();
995        sqlx::raw_sql(&sql).execute(&pool).await.unwrap();
996        let (id, title): (i64, String) = sqlx::query_as("SELECT id, title FROM tasks")
997            .fetch_one(&pool)
998            .await
999            .unwrap();
1000        assert_eq!(title, "existing");
1001        sqlx::query(
1002            "INSERT INTO task_dependencies (task_id, depends_on_repo, depends_on_number, source)
1003             VALUES (?, 'owner/repo', 1, 'cli')",
1004        )
1005        .bind(id)
1006        .execute(&pool)
1007        .await
1008        .unwrap();
1009        let (count,): (i64,) = sqlx::query_as("SELECT COUNT(*) FROM task_dependencies")
1010            .fetch_one(&pool)
1011            .await
1012            .unwrap();
1013        assert_eq!(count, 1);
1014    }
1015}
1016
1017/// The SQL behind each replicated dependency write
1018pub(crate) mod apply {
1019    use super::*;
1020
1021    pub(crate) async fn move_repo(
1022        conn: &mut SqliteConnection,
1023        old: &str,
1024        new: &str,
1025    ) -> anyhow::Result<u64> {
1026        let existing: i64 = sqlx::query_scalar("SELECT COUNT(*) FROM tasks WHERE repo = ?")
1027            .bind(new)
1028            .fetch_one(&mut *conn)
1029            .await?;
1030        if existing > 0 {
1031            anyhow::bail!("{new} already has {existing} task(s), refusing to move {old} onto it");
1032        }
1033        let moved = sqlx::query("UPDATE tasks SET repo = ? WHERE repo = ?")
1034            .bind(new)
1035            .bind(old)
1036            .execute(&mut *conn)
1037            .await?
1038            .rows_affected();
1039        for sql in [
1040            "UPDATE task_dependencies SET depends_on_repo = ? WHERE depends_on_repo = ?",
1041            "UPDATE task_metrics SET repo = ? WHERE repo = ?",
1042            "UPDATE job_state SET repo = ? WHERE repo = ?",
1043            "UPDATE channel_subscriptions SET repo = ? WHERE repo = ?",
1044        ] {
1045            sqlx::query(sql)
1046                .bind(new)
1047                .bind(old)
1048                .execute(&mut *conn)
1049                .await?;
1050        }
1051        move_repo_kv(&mut *conn, old, new).await?;
1052        Ok(moved)
1053    }
1054
1055    /**
1056     * Rename or drop the KV keys that embed a repo slug. Matching is by exact segment, so a
1057     * repo whose name merely starts with `old` keeps its keys, which a `LIKE '%old%'` would
1058     * have taken. Snapshot, read-at and batch-failure keys are dropped because the next
1059     * refresh rebuilds them. Notice, cycle-comment, approved-sha and ingest-cursor keys are
1060     * renamed, because dropping one posts its comment again or re-ingests from the start
1061     **/
1062    async fn move_repo_kv(conn: &mut SqliteConnection, old: &str, new: &str) -> anyhow::Result<()> {
1063        for prefix in [
1064            format!("dep_blocker:{old}#"),
1065            format!("dep_blocker_read_at:{old}#"),
1066        ] {
1067            sqlx::query("DELETE FROM kv WHERE substr(key, 1, ?) = ?")
1068                .bind(prefix_len(&prefix)?)
1069                .bind(&prefix)
1070                .execute(&mut *conn)
1071                .await?;
1072        }
1073        sqlx::query("DELETE FROM kv WHERE key = ?")
1074            .bind(format!("review_batch_fail:{old}"))
1075            .execute(&mut *conn)
1076            .await?;
1077        for (from, to) in [
1078            (
1079                format!("dep_cycle_comment:{old}#"),
1080                format!("dep_cycle_comment:{new}#"),
1081            ),
1082            (
1083                format!("review_approved_sha:{old}#"),
1084                format!("review_approved_sha:{new}#"),
1085            ),
1086        ] {
1087            sqlx::query(
1088                "UPDATE OR REPLACE kv SET key = ? || substr(key, ?) WHERE substr(key, 1, ?) = ?",
1089            )
1090            .bind(&to)
1091            .bind(prefix_len(&from)? + 1)
1092            .bind(prefix_len(&from)?)
1093            .bind(&from)
1094            .execute(&mut *conn)
1095            .await?;
1096        }
1097        // the repo sits mid-key, after the task id, and `:{repo}#` occurs once
1098        sqlx::query(
1099            "UPDATE OR REPLACE kv SET key = replace(key, ?, ?)
1100             WHERE key GLOB 'dep_blocker_notice:*' AND instr(key, ?) > 0",
1101        )
1102        .bind(format!(":{old}#"))
1103        .bind(format!(":{new}#"))
1104        .bind(format!(":{old}#"))
1105        .execute(&mut *conn)
1106        .await?;
1107        sqlx::query("UPDATE OR REPLACE kv SET key = ? WHERE key = ?")
1108            .bind(format!("issues_last_ingested:{new}"))
1109            .bind(format!("issues_last_ingested:{old}"))
1110            .execute(&mut *conn)
1111            .await?;
1112        Ok(())
1113    }
1114
1115    /// `substr` and `length` count characters, so the prefix length must too
1116    fn prefix_len(prefix: &str) -> anyhow::Result<i64> {
1117        Ok(i64::try_from(prefix.chars().count())?)
1118    }
1119
1120    pub(crate) async fn replace_dependencies(
1121        conn: &mut SqliteConnection,
1122        task_id: i64,
1123        rows: &[NewDependency],
1124    ) -> anyhow::Result<()> {
1125        sqlx::query("DELETE FROM task_dependencies WHERE task_id = ?")
1126            .bind(task_id)
1127            .execute(&mut *conn)
1128            .await?;
1129        for row in rows {
1130            if row.number <= 0 || row.repo.is_empty() {
1131                continue;
1132            }
1133            sqlx::query(
1134                "INSERT INTO task_dependencies
1135                 (task_id, depends_on_repo, depends_on_number, source)
1136                 VALUES (?, ?, ?, ?)",
1137            )
1138            .bind(task_id)
1139            .bind(&row.repo)
1140            .bind(row.number)
1141            .bind(row.source.as_str())
1142            .execute(&mut *conn)
1143            .await?;
1144        }
1145        sqlx::query(
1146            "UPDATE task_dependencies
1147             SET depends_on_task_id = (
1148                 SELECT id FROM tasks
1149                 WHERE tasks.repo = task_dependencies.depends_on_repo
1150                   AND tasks.external_id = CAST(task_dependencies.depends_on_number AS TEXT)
1151                 LIMIT 1
1152             )
1153             WHERE task_id = ?",
1154        )
1155        .bind(task_id)
1156        .execute(&mut *conn)
1157        .await?;
1158        Ok(())
1159    }
1160
1161    pub(crate) async fn link_known_task(
1162        conn: &mut SqliteConnection,
1163        task_id: i64,
1164        repo: &str,
1165        number: i64,
1166    ) -> anyhow::Result<()> {
1167        sqlx::query(
1168            "UPDATE task_dependencies
1169             SET depends_on_task_id = ?
1170             WHERE depends_on_repo = ? AND depends_on_number = ?",
1171        )
1172        .bind(task_id)
1173        .bind(repo)
1174        .bind(number)
1175        .execute(conn)
1176        .await?;
1177        Ok(())
1178    }
1179
1180    pub(crate) async fn block_claim(
1181        conn: &mut SqliteConnection,
1182        now: &str,
1183        task_id: i64,
1184    ) -> anyhow::Result<Option<Vec<String>>> {
1185        // sqlx 0.8.1 starts a deferred transaction. This write takes the reserved
1186        // lock before the read, so a second claim waits and then sees this one
1187        sqlx::query("UPDATE tasks SET updated_at = updated_at WHERE id = ?")
1188            .bind(task_id)
1189            .execute(&mut *conn)
1190            .await
1191            .context("lock task for dependency claim")?;
1192        let row = sqlx::query("SELECT status, agent, model, repo, body FROM tasks WHERE id = ?")
1193            .bind(task_id)
1194            .fetch_optional(&mut *conn)
1195            .await?;
1196        let Some(row) = row else {
1197            return Ok(None);
1198        };
1199        let status: String = row.try_get("status")?;
1200        let agent: Option<String> = row.try_get("agent")?;
1201        let model: Option<String> = row.try_get("model")?;
1202        let repo: String = row.try_get("repo")?;
1203        let body: String = row.try_get("body").unwrap_or_default();
1204        let views = load_blockers(conn, task_id, &repo).await?;
1205        let mut unmet: Vec<String> = views
1206            .iter()
1207            .filter(|view| !view.satisfied)
1208            .map(|view| view.label.clone())
1209            .collect();
1210        for id in blocked_external_from_body(&body) {
1211            if !unmet.iter().any(|have| have == &id) {
1212                unmet.push(id);
1213            }
1214        }
1215
1216        let holdable = matches!(
1217            status.as_str(),
1218            "new" | "routed" | "waiting" | "blocked" | "needs_review"
1219        );
1220        if unmet.is_empty() {
1221            if status == "waiting" {
1222                sqlx::query(
1223                    "UPDATE tasks
1224                     SET status = 'new', block_reason = NULL, branch = '', worktree = '',
1225                         updated_at = ?
1226                     WHERE id = ?",
1227                )
1228                .bind(now)
1229                .bind(task_id)
1230                .execute(&mut *conn)
1231                .await?;
1232                sqlx::query(
1233                    "INSERT INTO kv (key, value, updated_at)
1234                     VALUES (?, '1', ?)
1235                     ON CONFLICT(key) DO UPDATE
1236                       SET value = excluded.value, updated_at = excluded.updated_at",
1237                )
1238                .bind(worktree_refresh_key(task_id))
1239                .bind(now)
1240                .execute(&mut *conn)
1241                .await?;
1242                crate::store::tasks::apply::activity(
1243                    conn,
1244                    task_id,
1245                    "status_change",
1246                    Some("waiting"),
1247                    Some("new"),
1248                    agent.as_deref(),
1249                    model.as_deref(),
1250                    None,
1251                )
1252                .await?;
1253            }
1254            return Ok(None);
1255        }
1256
1257        if holdable && status != "waiting" {
1258            sqlx::query(
1259                "UPDATE tasks
1260                 SET status = 'waiting', branch = '', worktree = '', block_reason = NULL,
1261                     updated_at = ?
1262                 WHERE id = ?",
1263            )
1264            .bind(now)
1265            .bind(task_id)
1266            .execute(&mut *conn)
1267            .await?;
1268            crate::store::tasks::apply::activity(
1269                conn,
1270                task_id,
1271                "status_change",
1272                Some(&status),
1273                Some("waiting"),
1274                agent.as_deref(),
1275                model.as_deref(),
1276                None,
1277            )
1278            .await?;
1279        }
1280        Ok(Some(unmet))
1281    }
1282}
orch/store/dependencies.rs

orch/store/
dependencies.rs
