runner_updates.rs 91.0%

101 of 111 lines run, 29 of 62 functions · run #474 on main · run missed

//! Runner updates (oxygenhub#322): the policy at each level and the test
//! groups runners are put in.

use super::{now_ms, Pool};
use sqlx::FromRow;

/// One level's settings; `None` inherits from the next level out.
#[derive(Clone, Debug, Default, PartialEq, Eq, FromRow)]
pub struct Policy {
    pub scope: String,
    pub key: String,
    pub updates: Option<bool>,
    pub channel: Option<String>,
    pub window_from: Option<i64>,
    pub window_to: Option<i64>,
    pub dry_run: Option<bool>,
}

impl Policy {
    /// Nothing set: the level inherits everything.
    pub fn is_empty(&self) -> bool {
        self.updates.is_none() && self.channel.is_none() && self.window_from.is_none() && self.dry_run.is_none()
    }
}

#[derive(Clone, Debug, PartialEq, Eq, FromRow)]
pub struct Group {
    pub id: i64,
    /// Empty: a group of the site's runners.
    pub owner: String,
    pub name: String,
    /// A share of the runners in reach, by a stable hash of their ids;
    /// `None`: only the chosen ones.
    pub percent: Option<i64>,
    pub created_ms: i64,
}

pub async fn policies(pool: &Pool) -> sqlx::Result<Vec<Policy>> {
    sqlx::query_as("SELECT * FROM runner_update_policies").fetch_all(pool).await
}

pub async fn policy(pool: &Pool, scope: &str, key: &str) -> sqlx::Result<Policy> {
    let p: Option<Policy> = sqlx::query_as("SELECT * FROM runner_update_policies WHERE scope = $1 AND key = $2")
        .bind(scope)
        .bind(key)
        .fetch_optional(pool)
        .await?;
    Ok(p.unwrap_or(Policy { scope: scope.into(), key: key.into(), ..Policy::default() }))
}

/// Set a level's settings; one with nothing set is removed.
pub async fn set_policy(pool: &Pool, p: &Policy) -> sqlx::Result<()> {
    sqlx::query("DELETE FROM runner_update_policies WHERE scope = $1 AND key = $2").bind(&p.scope).bind(&p.key).execute(pool).await?;
    if p.is_empty() {
        return Ok(());
    }
    sqlx::query(
        "INSERT INTO runner_update_policies (scope, key, updates, channel, window_from, window_to, dry_run)
         VALUES ($1, $2, $3, $4, $5, $6, $7)",
    )
    .bind(&p.scope)
    .bind(&p.key)
    .bind(p.updates)
    .bind(&p.channel)
    .bind(p.window_from)
    .bind(p.window_to)
    .bind(p.dry_run)
    .execute(pool)
    .await?;
    Ok(())
}

pub async fn groups(pool: &Pool) -> sqlx::Result<Vec<Group>> {
    sqlx::query_as("SELECT * FROM runner_groups ORDER BY owner, name").fetch_all(pool).await
}

pub async fn group(pool: &Pool, id: i64) -> sqlx::Result<Option<Group>> {
    sqlx::query_as("SELECT * FROM runner_groups WHERE id = $1").bind(id).fetch_optional(pool).await
}

pub async fn create_group(pool: &Pool, owner: &str, name: &str, percent: Option<i64>) -> sqlx::Result<Group> {
    sqlx::query_as("INSERT INTO runner_groups (owner, name, percent, created_ms) VALUES ($1, $2, $3, $4) RETURNING *")
        .bind(owner)
        .bind(name)
        .bind(percent)
        .bind(now_ms())
        .fetch_one(pool)
        .await
}

pub async fn set_percent(pool: &Pool, id: i64, percent: Option<i64>) -> sqlx::Result<()> {
    sqlx::query("UPDATE runner_groups SET percent = $1 WHERE id = $2").bind(percent).bind(id).execute(pool).await?;
    Ok(())
}

/// Remove a group, its members and its policy.
pub async fn delete_group(pool: &Pool, id: i64) -> sqlx::Result<()> {
    sqlx::query("DELETE FROM runner_group_members WHERE group_id = $1").bind(id).execute(pool).await?;
    sqlx::query("DELETE FROM runner_update_policies WHERE scope = 'group' AND key = $1").bind(id.to_string()).execute(pool).await?;
    sqlx::query("DELETE FROM runner_groups WHERE id = $1").bind(id).execute(pool).await?;
    Ok(())
}

/// Every chosen member, as (group, runner).
pub async fn members(pool: &Pool) -> sqlx::Result<Vec<(i64, i64)>> {
    sqlx::query_as("SELECT group_id, runner_id FROM runner_group_members").fetch_all(pool).await
}

/// Make `runners` the group's chosen members.
pub async fn set_members(pool: &Pool, group: i64, runners: &[i64]) -> sqlx::Result<()> {
    sqlx::query("DELETE FROM runner_group_members WHERE group_id = $1").bind(group).execute(pool).await?;
    for r in runners {
        sqlx::query("INSERT INTO runner_group_members (group_id, runner_id) VALUES ($1, $2)").bind(group).bind(r).execute(pool).await?;
    }
    Ok(())
}

/// What a runner said about its update.
pub async fn report(pool: &Pool, runner: i64, note: &str) -> sqlx::Result<()> {
    sqlx::query("UPDATE flow_runners SET update_note = $1, update_ms = $2 WHERE id = $3")
        .bind(note)
        .bind(now_ms())
        .bind(runner)
        .execute(pool)
        .await?;
    Ok(())
}

/// A build held back from every runner (#326).
#[derive(Clone, Debug, PartialEq, Eq, FromRow)]
pub struct Hold {
    pub version: String,
    /// The runner it rolled back on.
    pub runner: String,
    pub why: String,
    pub at_ms: i64,
}

/// Hold `version` back, the first time it rolls back.
pub async fn hold(pool: &Pool, version: &str, runner: &str, why: &str) -> sqlx::Result<()> {
    sqlx::query("INSERT INTO runner_update_holds (version, runner, why, at_ms) VALUES ($1, $2, $3, $4) ON CONFLICT (version) DO NOTHING")
        .bind(version)
        .bind(runner)
        .bind(why)
        .bind(now_ms())
        .execute(pool)
        .await?;
    Ok(())
}

pub async fn holds(pool: &Pool) -> sqlx::Result<Vec<Hold>> {
    sqlx::query_as("SELECT * FROM runner_update_holds ORDER BY at_ms DESC").fetch_all(pool).await
}

pub async fn held(pool: &Pool, version: &str) -> sqlx::Result<Option<Hold>> {
    sqlx::query_as("SELECT * FROM runner_update_holds WHERE version = $1").bind(version).fetch_optional(pool).await
}

/// Lift the hold on `version`; whether there was one.
pub async fn lift(pool: &Pool, version: &str) -> sqlx::Result<bool> {
    Ok(sqlx::query("DELETE FROM runner_update_holds WHERE version = $1").bind(version).execute(pool).await?.rows_affected() > 0)
}

1		//! Runner updates (oxygenhub#322): the policy at each level and the test
2		//! groups runners are put in.
3
4		use super::{now_ms, Pool};
5		use sqlx::FromRow;
6
7		/// One level's settings; `None` inherits from the next level out.
8		#[derive(Clone, Debug, Default, PartialEq, Eq, FromRow)]
9		pub struct Policy {
10		pub scope: String,
11		pub key: String,
12		pub updates: Option<bool>,
13		pub channel: Option<String>,
14		pub window_from: Option<i64>,
15		pub window_to: Option<i64>,
16		pub dry_run: Option<bool>,
17		}
18
19		impl Policy {
20		/// Nothing set: the level inherits everything.
21	7×	pub fn is_empty(&self) -> bool {
22	7×	self.updates.is_none() && self.channel.is_none() && self.window_from.is_none() && self.dry_run.is_none()
23	7×	}
24		}
25
26		#[derive(Clone, Debug, PartialEq, Eq, FromRow)]
27		pub struct Group {
28		pub id: i64,
29		/// Empty: a group of the site's runners.
30		pub owner: String,
31		pub name: String,
32		/// A share of the runners in reach, by a stable hash of their ids;
33		/// `None`: only the chosen ones.
34		pub percent: Option<i64>,
35		pub created_ms: i64,
36		}
37
38	60×	pub async fn policies(pool: &Pool) -> sqlx::Result<Vec<Policy>> {
39	60×	sqlx::query_as("SELECT * FROM runner_update_policies").fetch_all(pool).await
40	60×	}
41
42	0×	pub async fn policy(pool: &Pool, scope: &str, key: &str) -> sqlx::Result<Policy> {
43	0×	let p: Option<Policy> = sqlx::query_as("SELECT * FROM runner_update_policies WHERE scope = $1 AND key = $2")
44	0×	.bind(scope)
45	0×	.bind(key)
46	0×	.fetch_optional(pool)
47	0×	.await?;
48	0×	Ok(p.unwrap_or(Policy { scope: scope.into(), key: key.into(), ..Policy::default() }))
49	0×	}
50
51		/// Set a level's settings; one with nothing set is removed.
52	7×	pub async fn set_policy(pool: &Pool, p: &Policy) -> sqlx::Result<()> {
53	7×	sqlx::query("DELETE FROM runner_update_policies WHERE scope = $1 AND key = $2").bind(&p.scope).bind(&p.key).execute(pool).await?;
54	7×	if p.is_empty() {
55	0×	return Ok(());
56	7×	}
57	7×	sqlx::query(
58	7×	"INSERT INTO runner_update_policies (scope, key, updates, channel, window_from, window_to, dry_run)
59	7×	VALUES ($1, $2, $3, $4, $5, $6, $7)",
60	7×	)
61	7×	.bind(&p.scope)
62	7×	.bind(&p.key)
63	7×	.bind(p.updates)
64	7×	.bind(&p.channel)
65	7×	.bind(p.window_from)
66	7×	.bind(p.window_to)
67	7×	.bind(p.dry_run)
68	7×	.execute(pool)
69	7×	.await?;
70	7×	Ok(())
71	7×	}
72
73	64×	pub async fn groups(pool: &Pool) -> sqlx::Result<Vec<Group>> {
74	64×	sqlx::query_as("SELECT * FROM runner_groups ORDER BY owner, name").fetch_all(pool).await
75	64×	}
76
77	7×	pub async fn group(pool: &Pool, id: i64) -> sqlx::Result<Option<Group>> {
78	7×	sqlx::query_as("SELECT * FROM runner_groups WHERE id = $1").bind(id).fetch_optional(pool).await
79	7×	}
80
81	3×	pub async fn create_group(pool: &Pool, owner: &str, name: &str, percent: Option<i64>) -> sqlx::Result<Group> {
82	3×	sqlx::query_as("INSERT INTO runner_groups (owner, name, percent, created_ms) VALUES ($1, $2, $3, $4) RETURNING *")
83	3×	.bind(owner)
84	3×	.bind(name)
85	3×	.bind(percent)
86	3×	.bind(now_ms())
87	3×	.fetch_one(pool)
88	3×	.await
89	3×	}
90
91	1×	pub async fn set_percent(pool: &Pool, id: i64, percent: Option<i64>) -> sqlx::Result<()> {
92	1×	sqlx::query("UPDATE runner_groups SET percent = $1 WHERE id = $2").bind(percent).bind(id).execute(pool).await?;
93	1×	Ok(())
94	1×	}
95
96		/// Remove a group, its members and its policy.
97	1×	pub async fn delete_group(pool: &Pool, id: i64) -> sqlx::Result<()> {
98	1×	sqlx::query("DELETE FROM runner_group_members WHERE group_id = $1").bind(id).execute(pool).await?;
99	1×	sqlx::query("DELETE FROM runner_update_policies WHERE scope = 'group' AND key = $1").bind(id.to_string()).execute(pool).await?;
100	1×	sqlx::query("DELETE FROM runner_groups WHERE id = $1").bind(id).execute(pool).await?;
101	1×	Ok(())
102	1×	}
103
104		/// Every chosen member, as (group, runner).
105	60×	pub async fn members(pool: &Pool) -> sqlx::Result<Vec<(i64, i64)>> {
106	60×	sqlx::query_as("SELECT group_id, runner_id FROM runner_group_members").fetch_all(pool).await
107	60×	}
108
109		/// Make `runners` the group's chosen members.
110	3×	pub async fn set_members(pool: &Pool, group: i64, runners: &[i64]) -> sqlx::Result<()> {
111	3×	sqlx::query("DELETE FROM runner_group_members WHERE group_id = $1").bind(group).execute(pool).await?;
112	3×	for r in runners {
113	2×	sqlx::query("INSERT INTO runner_group_members (group_id, runner_id) VALUES ($1, $2)").bind(group).bind(r).execute(pool).await?;
114		}
115	3×	Ok(())
116	3×	}
117
118		/// What a runner said about its update.
119	2×	pub async fn report(pool: &Pool, runner: i64, note: &str) -> sqlx::Result<()> {
120	2×	sqlx::query("UPDATE flow_runners SET update_note = $1, update_ms = $2 WHERE id = $3")
121	2×	.bind(note)
122	2×	.bind(now_ms())
123	2×	.bind(runner)
124	2×	.execute(pool)
125	2×	.await?;
126	2×	Ok(())
127	2×	}
128
129		/// A build held back from every runner (#326).
130		#[derive(Clone, Debug, PartialEq, Eq, FromRow)]
131		pub struct Hold {
132		pub version: String,
133		/// The runner it rolled back on.
134		pub runner: String,
135		pub why: String,
136		pub at_ms: i64,
137		}
138
139		/// Hold `version` back, the first time it rolls back.
140	1×	pub async fn hold(pool: &Pool, version: &str, runner: &str, why: &str) -> sqlx::Result<()> {
141	1×	sqlx::query("INSERT INTO runner_update_holds (version, runner, why, at_ms) VALUES ($1, $2, $3, $4) ON CONFLICT (version) DO NOTHING")
142	1×	.bind(version)
143	1×	.bind(runner)
144	1×	.bind(why)
145	1×	.bind(now_ms())
146	1×	.execute(pool)
147	1×	.await?;
148	1×	Ok(())
149	1×	}
150
151	17×	pub async fn holds(pool: &Pool) -> sqlx::Result<Vec<Hold>> {
152	17×	sqlx::query_as("SELECT * FROM runner_update_holds ORDER BY at_ms DESC").fetch_all(pool).await
153	17×	}
154
155	17×	pub async fn held(pool: &Pool, version: &str) -> sqlx::Result<Option<Hold>> {
156	17×	sqlx::query_as("SELECT * FROM runner_update_holds WHERE version = $1").bind(version).fetch_optional(pool).await
157	17×	}
158
159		/// Lift the hold on `version`; whether there was one.
160	2×	pub async fn lift(pool: &Pool, version: &str) -> sqlx::Result<bool> {
161	2×	Ok(sqlx::query("DELETE FROM runner_update_holds WHERE version = $1").bind(version).execute(pool).await?.rows_affected() > 0)
162	2×	}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

Coverage / crates/hub/src/db/runner_updates.rs 91.0%

Keyboard