Skip to content
Sign up

flagon-io/g1t

public
0
243 linesCodeBlameRaw
1//! Coordinating work between agents and people on one repository
2//! (`g1t_contracts::work::coordination`): sessions of their own harnesses
3//! (sessions.rs), advisory claims on issues (claims.rs), what an issue is
4//! blocked by (dependencies.rs), pull requests that start changing the same
5//! files (overlaps.rs), and the one call that gathers it all for an issue
6//! (context.rs).
7//!
8//! Nothing here is a lock and nothing waits on a harness: sessions and
9//! claims that went quiet are found when the repository's coordination is
10//! next touched (`sweep`), and announced then as ended or expired.
11
12mod claims;
13mod context;
14mod dependencies;
15mod overlaps;
16mod sessions;
17#[cfg(test)]
18mod tests;
19
20use g1t_contracts::credentials::Principal;
21use g1t_contracts::repos::{Repo, RepoPath};
22use g1t_contracts::time::{parse_rfc3339, rfc3339};
23use g1t_contracts::work::*;
24use g1t_contracts::{FailureCode, Outcome, User, Viewer};
25use g1t_kit::now_ms;
26use serde::Deserialize;
27use worker::Result;
28
29use crate::Work;
30
31/// Unwraps an `Outcome`, returning its failure from the enclosing method.
32macro_rules! unwrap_or_fail {
33 ($outcome:expr) => {
34 match $outcome {
35 g1t_contracts::Outcome::Ok(value) => value,
36 g1t_contracts::Outcome::Fail(failure) => return Ok(g1t_contracts::Outcome::Fail(failure)),
37 }
38 };
39}
40pub(crate) use unwrap_or_fail;
41
42const SESSION_COLUMNS: &str = "id, repo_id, issue, pull, branch, actor_id, actor_name, harness, harness_session_id,
43 title, state, head_commit, started_at, last_seen_at, ended_at, end_reason";
44
45const CLAIM_COLUMNS: &str = "id, repo_id, issue, session_id, actor_id, actor_name, harness, note, state,
46 ttl_minutes, created_at, expires_at, ended_at, end_reason, handoff_to";
47
48#[derive(Deserialize)]
49pub(crate) struct SessionRow {
50 id: String,
51 repo_id: String,
52 issue: Option<u32>,
53 pull: Option<u32>,
54 branch: Option<String>,
55 actor_id: String,
56 actor_name: String,
57 harness: String,
58 harness_session_id: Option<String>,
59 title: Option<String>,
60 state: String,
61 head_commit: Option<String>,
62 started_at: String,
63 last_seen_at: String,
64 ended_at: Option<String>,
65 end_reason: Option<String>,
66}
67
68impl SessionRow {
69 /// The session as it reads at `now`.
70 pub(crate) fn at(self, now: u64) -> Session {
71 let last_seen = parse_rfc3339(&self.last_seen_at).unwrap_or(now);
72 Session {
73 state: SessionState::at(self.state == "ended", last_seen, now),
74 id: self.id,
75 repo_id: self.repo_id,
76 issue: self.issue,
77 pull: self.pull,
78 branch: self.branch,
79 actor: Principal { id: self.actor_id, username: self.actor_name },
80 harness: Harness::parse(&self.harness).unwrap_or(Harness::Other),
81 harness_session_id: self.harness_session_id,
82 title: self.title,
83 head_commit: self.head_commit,
84 started_at: self.started_at,
85 last_seen_at: self.last_seen_at,
86 ended_at: self.ended_at,
87 end_reason: self.end_reason,
88 }
89 }
90}
91
92#[derive(Deserialize)]
93pub(crate) struct ClaimRow {
94 id: String,
95 repo_id: String,
96 issue: u32,
97 session_id: Option<String>,
98 actor_id: String,
99 actor_name: String,
100 harness: Option<String>,
101 note: Option<String>,
102 state: String,
103 #[allow(dead_code)]
104 ttl_minutes: u32,
105 created_at: String,
106 expires_at: String,
107 ended_at: Option<String>,
108 end_reason: Option<String>,
109 handoff_to: Option<String>,
110}
111
112impl From<ClaimRow> for Claim {
113 fn from(row: ClaimRow) -> Self {
114 Claim {
115 id: row.id,
116 repo_id: row.repo_id,
117 issue: row.issue,
118 session_id: row.session_id,
119 actor: Principal { id: row.actor_id, username: row.actor_name },
120 harness: row.harness.as_deref().and_then(Harness::parse),
121 note: row.note,
122 state: ClaimState::parse(&row.state).unwrap_or(ClaimState::Expired),
123 created_at: row.created_at,
124 expires_at: row.expires_at,
125 ended_at: row.ended_at,
126 end_reason: row.end_reason,
127 handoff_to: row.handoff_to,
128 }
129 }
130}
131
132/// Text as given, trimmed and cut to `limit` characters; none when empty.
133pub(crate) fn tidy(text: Option<&str>, limit: usize) -> Option<String> {
134 let text = text?.trim();
135 (!text.is_empty()).then(|| text.chars().take(limit).collect())
136}
137
138/// Ends a repository's (`?1`) sessions last seen at or before `?2` as
139/// `timed_out`, as of when they were last seen.
140pub(crate) fn time_out_sessions_sql() -> String {
141 format!(
142 "UPDATE sessions SET state = 'ended', ended_at = last_seen_at, end_reason = 'timed_out'
143 WHERE repo_id = ?1 AND state = 'active' AND last_seen_at <= ?2
144 RETURNING {SESSION_COLUMNS}"
145 )
146}
147
148/// Expires a repository's (`?1`) active claims due at or before `?2`.
149pub(crate) fn expire_claims_sql() -> String {
150 format!(
151 "UPDATE claims SET state = 'expired', ended_at = expires_at, end_reason = 'expired'
152 WHERE repo_id = ?1 AND state = 'active' AND expires_at <= ?2
153 RETURNING {CLAIM_COLUMNS}"
154 )
155}
156
157/// RFC 3339 for `minutes` after `now`.
158pub(crate) fn after_minutes(now: u64, minutes: u32) -> String {
159 rfc3339(now + u64::from(minutes) * 60_000)
160}
161
162impl Work {
163 /// The repository as `viewer` sees it, for reading its coordination.
164 pub(crate) async fn coordination_repo(&self, path: &RepoPath, viewer: &Viewer) -> Result<Outcome<Repo>> {
165 let repo = unwrap_or_fail!(self.repo(path, viewer).await?);
166 self.sweep(&repo.id).await?;
167 Ok(Outcome::Ok(repo))
168 }
169
170 /// The repository, if `actor` may take part in it: a confirmed email
171 /// address, a repository that is not archived, and a role that may
172 /// open issues and pull requests.
173 pub(crate) async fn participating_repo(&self, path: &RepoPath, actor: &User) -> Result<Outcome<Repo>> {
174 if !actor.verified {
175 return Ok(Outcome::fail(FailureCode::Forbidden, crate::UNVERIFIED));
176 }
177 let repo = unwrap_or_fail!(self.repo(path, &Some(actor.clone())).await?);
178 unwrap_or_fail!(crate::retired::writable(&repo));
179 unwrap_or_fail!(crate::allowed(Some(actor), &repo, g1t_contracts::access::Capability::Participate));
180 self.sweep(&repo.id).await?;
181 Ok(Outcome::Ok(repo))
182 }
183
184 /// Ends the repository's sessions that have gone two hours without a
185 /// heartbeat (`timed_out`) and expires its claims past their time, and
186 /// announces each. Run before the repository's coordination is read or
187 /// changed, so that nobody sees a stale one as live.
188 pub(crate) async fn sweep(&self, repo_id: &str) -> Result<()> {
189 let now = now_ms();
190 let quiet_since = rfc3339(now.saturating_sub(SESSION_TIMEOUT_MS));
191 let at = rfc3339(now);
192 let results = self
193 .db
194 .batch(vec![
195 self.db.prepare(time_out_sessions_sql()).bind(&[repo_id.into(), quiet_since.as_str().into()])?,
196 self.db.prepare(expire_claims_sql()).bind(&[repo_id.into(), at.as_str().into()])?,
197 ])
198 .await?;
199 let mut results = results.into_iter();
200 let ended: Vec<SessionRow> = match results.next() {
201 Some(result) => result.results()?,
202 None => Vec::new(),
203 };
204 let expired: Vec<ClaimRow> = match results.next() {
205 Some(result) => result.results()?,
206 None => Vec::new(),
207 };
208 for row in ended {
209 let session = row.at(now);
210 self.publish_as("session.ended", repo_id, None, SessionEvent::of(&session)).await?;
211 }
212 for row in expired {
213 let claim = Claim::from(row);
214 self.publish_as("work.claim_expired", repo_id, None, ClaimEvent::of(&claim)).await?;
215 }
216 Ok(())
217 }
218
219 pub(crate) async fn session_row(&self, repo_id: &str, id: &str) -> Result<Option<SessionRow>> {
220 self.db
221 .prepare(format!("SELECT {SESSION_COLUMNS} FROM sessions WHERE repo_id = ? AND id = ?"))
222 .bind(&[repo_id.into(), id.into()])?
223 .first::<SessionRow>(None)
224 .await
225 }
226
227 /// Active claims on an issue, oldest first.
228 pub(crate) async fn active_claims(&self, repo_id: &str, issue: u32) -> Result<Vec<Claim>> {
229 Ok(self
230 .db
231 .prepare(format!(
232 "SELECT {CLAIM_COLUMNS} FROM claims
233 WHERE repo_id = ? AND issue = ? AND state = 'active' ORDER BY created_at, id"
234 ))
235 .bind(&[repo_id.into(), issue.into()])?
236 .all()
237 .await?
238 .results::<ClaimRow>()?
239 .into_iter()
240 .map(Claim::from)
241 .collect())
242 }
243}