Skip to content
Sign up

flagon-io/g1t

public
0

Commit

Self-hosted runners' polls write that they were seen once a minute, and find queued jobs by an index (actions 0010)

syntaqxcommitted Parent5cb42efBrowse files
2 files+47−130/2 viewed
+5−0
1+-- What a self-hosted runner can take, asked on every poll: queued jobs with
2+-- labels in its workspace, matched without regard to case. By the
3+-- lowercased namespace, so the match is a range of the index rather than
4+-- every labelled job.
5+CREATE INDEX IF NOT EXISTS jobs_self_hosted_lower ON jobs (lower(namespace), status) WHERE labels IS NOT NULL;
+42−13
3636 const POLL_EVERY_MS: u64 = 2_500;
3737 /// A runner that has been offline this long is removed, as on GitHub.
3838 const FORGET_OFFLINE_MS: u64 = 14 * 24 * 60 * 60 * 1000;
39+/// How often a poll writes down that a runner, or the work it holds, is
40+/// still there. Polls come every twenty seconds or so; a write each time
41+/// would be most of the service's writes for nothing. Well inside
42+/// [`model::ONLINE_WITHIN_MS`] and [`crate::SILENT_MS`].
43+const SEEN_RESOLUTION_MS: u64 = 60 * 1000;
44+
45+/// Whether `seen` is old enough, at `at`, to be written down again.
46+fn seen_is_stale(seen: Option<&str>, at: u64) -> bool {
47+ seen.is_none_or(|seen| seen < rfc3339(at.saturating_sub(SEEN_RESOLUTION_MS)).as_str())
48+}
3949
4050 #[derive(Clone, Debug, Deserialize)]
4151 pub struct RunnerRow {
124134 pub runner_id: Option<String>,
125135 pub runner_name: Option<String>,
126136 pub started_at: Option<String>,
137+ #[serde(default)]
138+ pub seen_at: Option<String>,
127139 }
128140
129141 /// Who may do what with a set of runners: the workspace, and the
912924 for id in a.running.iter().take(10) {
913925 if let Some(job) = self.db.prepare("SELECT * FROM jobs WHERE id = ?").bind(&[id.as_str().into()])?.first::<JobRow>(None).await? {
914926 if job.status == "in_progress" && job.runner_id.as_deref() == Some(row.id.as_str()) {
915− self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[seen.as_str().into(), id.as_str().into()])?.run().await?;
927+ if seen_is_stale(job.seen_at.as_deref(), at) {
928+ self.db.prepare("UPDATE jobs SET seen_at = ? WHERE id = ?").bind(&[seen.as_str().into(), id.as_str().into()])?.run().await?;
929+ }
916930 active = Some((id.clone(), "workflow"));
917931 } else {
918932 cancel.push(id.clone());
921935 }
922936 match self.db.prepare("SELECT * FROM runner_tasks WHERE id = ?").bind(&[id.as_str().into()])?.first::<TaskRow>(None).await? {
923937 Some(task) if task.status == "in_progress" && task.runner_id.as_deref() == Some(row.id.as_str()) => {
924− self.db.prepare("UPDATE runner_tasks SET seen_at = ? WHERE id = ?").bind(&[seen.as_str().into(), id.as_str().into()])?.run().await?;
938+ if seen_is_stale(task.seen_at.as_deref(), at) {
939+ self.db.prepare("UPDATE runner_tasks SET seen_at = ? WHERE id = ?").bind(&[seen.as_str().into(), id.as_str().into()])?.run().await?;
940+ }
925941 active = Some((id.clone(), "agent"));
926942 }
927943 _ => cancel.push(id.clone()),
928944 }
929945 }
930− self.db
931− .prepare("UPDATE runners SET last_seen_at = ?, version = ?, work_id = ?, work_kind = ? WHERE id = ?")
932− .bind(&[
933− seen.as_str().into(),
934− (if a.version.is_empty() { row.version.clone() } else { a.version.chars().take(40).collect() }).into(),
935− optional(active.as_ref().map(|(id, _)| id.as_str())),
936− optional(active.as_ref().map(|(_, kind)| *kind)),
937− row.id.as_str().into(),
938− ])?
939− .run()
940− .await?;
946+ // Written when something changed, or once a minute to keep it online.
947+ let version: String = if a.version.is_empty() { row.version.clone() } else { a.version.chars().take(40).collect() };
948+ let work_id = active.as_ref().map(|(id, _)| id.as_str());
949+ let work_kind = active.as_ref().map(|(_, kind)| *kind);
950+ let changed = version != row.version || work_id != row.work_id.as_deref() || work_kind != row.work_kind.as_deref();
951+ if changed || seen_is_stale(row.last_seen_at.as_deref(), at) {
952+ self.db
953+ .prepare("UPDATE runners SET last_seen_at = ?, version = ?, work_id = ?, work_kind = ? WHERE id = ?")
954+ .bind(&[seen.as_str().into(), version.into(), optional(work_id), optional(work_kind), row.id.as_str().into()])?
955+ .run()
956+ .await?;
957+ }
941958 let mut poll = Poll { assignment: None, cancel, credential, removed: false };
942959 // Busy, or an ephemeral runner that has had its one job.
943960 if active.is_some() || !a.running.is_empty() {
15431560 }
15441561
15451562 #[test]
1563+ fn being_seen_is_written_once_a_minute() {
1564+ let at = 2_000_000_000_000;
1565+ assert!(seen_is_stale(None, at));
1566+ assert!(!seen_is_stale(Some(&rfc3339(at - 59_000)), at));
1567+ assert!(seen_is_stale(Some(&rfc3339(at - 61_000)), at));
1568+ // A runner written down at the last moment is still online until the
1569+ // next write, a poll later.
1570+ assert!(SEEN_RESOLUTION_MS + model::MAX_POLL_WAIT_MS < model::ONLINE_WITHIN_MS);
1571+ assert!(SEEN_RESOLUTION_MS < crate::SILENT_MS);
1572+ }
1573+
1574+ #[test]
15461575 fn a_group_lets_its_repositories_use_it() {
15471576 assert!(group(&[]).allows("acme/web"));
15481577 assert!(group(&["web", "api"]).allows("acme/WEB"));