Fix new issues after rebase

Author: jtgeibel

src/background_jobs.rs

@@ -58,7 +58,7 @@ impl Job {
         }
     }
 
-    pub fn enqueue(&self, conn: &PgConnection) -> Result<(), EnqueueError> {
+    pub fn enqueue(&self, conn: &mut PgConnection) -> Result<(), EnqueueError> {
         use crate::schema::background_jobs::dsl::*;
 
         let job_data = self.to_value()?;

src/controllers/github/secret_scanning.rs

@@ -155,14 +155,14 @@ fn alert_revoke_token(
     state: &AppState,
     alert: &GitHubSecretAlert,
 ) -> Result<GitHubSecretAlertFeedbackLabel, Box<dyn AppError>> {
-    let conn = state.db_write()?;
+    let conn = &mut *state.db_write()?;
 
     let hashed_token = SecureToken::hash(&alert.token);
 
     // Not using `ApiToken::find_by_api_token()` in order to preserve `last_used_at`
     let token = api_tokens::table
         .filter(api_tokens::token.eq(hashed_token))
-        .get_result::<ApiToken>(&*conn)
+        .get_result::<ApiToken>(conn)
         .optional()?;
 
     let Some(token) = token else {
@@ -180,7 +180,7 @@ fn alert_revoke_token(
 
     diesel::update(&token)
         .set(api_tokens::revoked.eq(true))
-        .execute(&*conn)?;
+        .execute(conn)?;
 
     warn!(
         token_id = %token.id, user_id = %token.user_id,
@@ -202,10 +202,10 @@ fn send_notification_email(
     alert: &GitHubSecretAlert,
     state: &AppState,
 ) -> anyhow::Result<()> {
-    let conn = state.db_read()?;
+    let conn = &mut state.db_read()?;
 
-    let user = User::find(&conn, token.user_id).context("Failed to find user")?;
-    let Some(email) = user.email(&conn)? else {
+    let user = User::find(conn, token.user_id).context("Failed to find user")?;
+    let Some(email) = user.email(conn)? else {
         return Err(anyhow!("No address found"));
     };
 

src/controllers/krate/metadata.rs

@@ -351,7 +351,7 @@ pub fn versions(req: &mut dyn RequestExt) -> EndpointResult {
 pub fn reverse_dependencies(req: &mut dyn RequestExt) -> EndpointResult {
     let pagination_options = PaginationOptions::builder().gather(req)?;
     let name = &req.params()["crate_id"];
-    let conn = &mut req.app().db_read()?;
+    let conn = &mut *req.app().db_read()?;
     let krate: Crate = Crate::by_name(name).first(conn)?;
     let (rev_deps, total) = krate.reverse_dependencies(conn, pagination_options)?;
     let rev_deps: Vec<_> = rev_deps

src/controllers/krate/publish.rs

@@ -69,7 +69,7 @@ pub fn publish(req: &mut dyn RequestExt) -> EndpointResult {
     // since a race condition there would only cause `publish-new` instead of
     // `publish-update` to be used.
     let existing_crate = Crate::by_name(&new_crate.name)
-        .first::<Crate>(&*conn)
+        .first::<Crate>(conn)
         .optional()?;
 
     let endpoint_scope = match existing_crate {
@@ -146,7 +146,7 @@ pub fn publish(req: &mut dyn RequestExt) -> EndpointResult {
         }
 
         if let Some(daily_version_limit) = app.config.new_version_rate_limit {
-            let published_today = count_versions_published_today(krate.id, &conn)?;
+            let published_today = count_versions_published_today(krate.id, conn)?;
             if published_today >= daily_version_limit as i64 {
                 return Err(cargo_err(
                     "You have published too many versions of this crate in the last 24 hours",
@@ -287,7 +287,7 @@ pub fn publish(req: &mut dyn RequestExt) -> EndpointResult {
 
 /// Counts the number of versions for `krate_id` that were published within
 /// the last 24 hours.
-fn count_versions_published_today(krate_id: i32, conn: &PgConnection) -> QueryResult<i64> {
+fn count_versions_published_today(krate_id: i32, conn: &mut PgConnection) -> QueryResult<i64> {
     use crate::schema::versions::dsl::*;
     use diesel::dsl::{now, IntervalDsl};
 

src/controllers/version/yank.rs

@@ -40,8 +40,8 @@ fn modify_yank(req: &mut dyn RequestExt, yanked: bool) -> EndpointResult {
         .for_crate(crate_name)
         .check(req)?;
 
-    let state = &mut *req.app();
-    let conn = &mut state.db_write()?;
+    let state = req.app();
+    let conn = &mut *state.db_write()?;
     let (version, krate) = version_and_crate(conn, crate_name, semver)?;
     let api_token_id = auth.api_token_id();
     let user = auth.user();

src/db.rs

@@ -25,15 +25,15 @@ pub enum DieselPool {
     Test(Arc<ReentrantMutex<PgConnection>>),
 }
 
-type Callback<'a> = &'a dyn Fn(&PgConnection) -> Result<(), Box<dyn Error>>;
+type Callback<'a> = &'a dyn Fn(&mut PgConnection) -> Result<(), Box<dyn Error>>;
 
 impl DieselPool {
     pub(crate) fn with_connection(&self, f: Callback<'_>) -> Result<(), Box<dyn Error>> {
         match self {
             DieselPool::Pool { pool, .. } | DieselPool::BackgroundJobPool { pool } => {
-                f(&*pool.get()?)
+                f(&mut *pool.get()?)
             }
-            DieselPool::Test(connection) => f(&connection.lock()),
+            DieselPool::Test(connection) => f(&mut connection.lock()),
         }
     }
 

src/models/token/scopes.rs

@@ -26,7 +26,7 @@ impl From<&EndpointScope> for &[u8] {
 }
 
 impl ToSql<Text, Pg> for EndpointScope {
-    fn to_sql<W: Write>(&self, out: &mut Output<'_, W, Pg>) -> serialize::Result {
+    fn to_sql(&self, out: &mut Output<'_, '_, Pg>) -> serialize::Result {
         out.write_all(self.into())?;
         Ok(IsNull::No)
     }
@@ -47,8 +47,9 @@ impl TryFrom<&[u8]> for EndpointScope {
 }
 
 impl FromSql<Text, Pg> for EndpointScope {
-    fn from_sql(bytes: Option<&[u8]>) -> deserialize::Result<Self> {
-        Ok(EndpointScope::try_from(not_none!(bytes))?)
+    fn from_sql(bytes: diesel::pg::PgValue<'_>) -> deserialize::Result<Self> {
+        let value = <String as FromSql<Text, Pg>>::from_sql(bytes)?;
+        Ok(EndpointScope::try_from(value.as_bytes())?)
     }
 }
 
@@ -82,15 +83,15 @@ impl TryFrom<String> for CrateScope {
 }
 
 impl FromSql<Text, Pg> for CrateScope {
-    fn from_sql(bytes: Option<&[u8]>) -> deserialize::Result<Self> {
+    fn from_sql(bytes: diesel::pg::PgValue<'_>) -> deserialize::Result<Self> {
         let value = <String as FromSql<Text, Pg>>::from_sql(bytes)?;
         Ok(CrateScope::try_from(value)?)
     }
 }
 
 impl ToSql<Text, Pg> for CrateScope {
-    fn to_sql<W: Write>(&self, out: &mut Output<'_, W, Pg>) -> serialize::Result {
-        <String as ToSql<Text, Pg>>::to_sql(&self.pattern, out)
+    fn to_sql(&self, out: &mut Output<'_, '_, Pg>) -> serialize::Result {
+        ToSql::<Text, Pg>::to_sql(&self.pattern, &mut out.reborrow())
     }
 }
 

src/swirl/runner.rs

@@ -140,7 +140,7 @@ impl Runner {
         // The connection may not be `Send` so we need to clone the pool instead
         let pool = self.connection_pool.clone();
         self.thread_pool.execute(move || {
-            let conn = &*match pool.get() {
+            let conn = &mut *match pool.get() {
                 Ok(conn) => conn,
                 Err(e) => {
                     // TODO: Review error handling and possibly drop all usage of `let _ =` in this file
@@ -149,7 +149,7 @@ impl Runner {
                 }
             };
 
-            let job_run_result = conn.transaction::<_, diesel::result::Error, _>(|| {
+            let job_run_result = conn.transaction::<_, diesel::result::Error, _>(|conn| {
                 let job = match storage::find_next_unlocked_job(conn).optional() {
                     Ok(Some(j)) => {
                         let _ = sender.send(Event::Working);
@@ -205,7 +205,7 @@ impl Runner {
     // FIXME: Only public for `src/tests/util/test_app.rs`
     pub fn check_for_failed_jobs(&self) -> Result<(), FailedJobsError> {
         self.wait_for_jobs()?;
-        let failed_jobs = storage::failed_job_count(&*self.connection()?)?;
+        let failed_jobs = storage::failed_job_count(&mut *self.connection()?)?;
         if failed_jobs == 0 {
             Ok(())
         } else {
@@ -298,7 +298,7 @@ mod tests {
 
         let remaining_jobs = background_jobs
             .count()
-            .get_result(&*runner.connection().unwrap());
+            .get_result(&mut *runner.connection().unwrap());
         assert_eq!(Ok(0), remaining_jobs);
     }
 
@@ -317,7 +317,7 @@ mod tests {
             Err("nope".into())
         });
 
-        let conn = &*runner.connection().unwrap();
+        let conn = &mut *runner.connection().unwrap();
         // Wait for the first thread to acquire the lock
         barrier2.0.wait();
         // We are intentionally not using `get_single_job` here.
@@ -357,7 +357,7 @@ mod tests {
             .find(job_id)
             .select(retries)
             .for_update()
-            .first::<i32>(&*runner.connection().unwrap())
+            .first::<i32>(&mut *runner.connection().unwrap())
             .unwrap();
         assert_eq!(1, tries);
     }
@@ -381,7 +381,7 @@ mod tests {
     impl<'a> Drop for TestGuard<'a> {
         fn drop(&mut self) {
             ::diesel::sql_query("TRUNCATE TABLE background_jobs")
-                .execute(&*runner().connection().unwrap())
+                .execute(&mut *runner().connection().unwrap())
                 .unwrap();
         }
     }
@@ -397,7 +397,7 @@ mod tests {
         ::diesel::insert_into(background_jobs)
             .values((job_type.eq("Foo"), data.eq(serde_json::json!(null))))
             .returning((id, job_type, data))
-            .get_result(&*runner.connection().unwrap())
+            .get_result(&mut *runner.connection().unwrap())
             .unwrap()
     }
 }

src/swirl/storage.rs

@@ -24,7 +24,7 @@ fn retriable() -> Box<dyn BoxableExpression<background_jobs::table, Pg, SqlType
 
 /// Finds the next job that is unlocked, and ready to be retried. If a row is
 /// found, it will be locked.
-pub(super) fn find_next_unlocked_job(conn: &PgConnection) -> QueryResult<BackgroundJob> {
+pub(super) fn find_next_unlocked_job(conn: &mut PgConnection) -> QueryResult<BackgroundJob> {
     use schema::background_jobs::dsl::*;
 
     background_jobs
@@ -37,7 +37,7 @@ pub(super) fn find_next_unlocked_job(conn: &PgConnection) -> QueryResult<Backgro
 }
 
 /// The number of jobs that have failed at least once
-pub(super) fn failed_job_count(conn: &PgConnection) -> QueryResult<i64> {
+pub(super) fn failed_job_count(conn: &mut PgConnection) -> QueryResult<i64> {
     use schema::background_jobs::dsl::*;
 
     background_jobs
@@ -47,7 +47,7 @@ pub(super) fn failed_job_count(conn: &PgConnection) -> QueryResult<i64> {
 }
 
 /// Deletes a job that has successfully completed running
-pub(super) fn delete_successful_job(conn: &PgConnection, job_id: i64) -> QueryResult<()> {
+pub(super) fn delete_successful_job(conn: &mut PgConnection, job_id: i64) -> QueryResult<()> {
     use schema::background_jobs::dsl::*;
 
     delete(background_jobs.find(job_id)).execute(conn)?;
@@ -58,7 +58,7 @@ pub(super) fn delete_successful_job(conn: &PgConnection, job_id: i64) -> QueryRe
 ///
 /// Ignores any database errors that may have occurred. If the DB has gone away,
 /// we assume that just trying again with a new connection will succeed.
-pub(super) fn update_failed_job(conn: &PgConnection, job_id: i64) {
+pub(super) fn update_failed_job(conn: &mut PgConnection, job_id: i64) {
     use schema::background_jobs::dsl::*;
 
     let _ = update(background_jobs.find(job_id))

src/worker/git.rs

@@ -14,7 +14,7 @@ use std::process::Command;
 #[instrument(skip_all, fields(krate.name = ?krate.name, krate.vers = ?krate.vers))]
 pub fn perform_index_add_crate(
     env: &Environment,
-    conn: &PgConnection,
+    conn: &mut PgConnection,
     krate: &Crate,
 ) -> Result<(), PerformError> {
     info!("Syncing git index to HTTP-based index");

src/worker/update_downloads.rs

@@ -40,8 +40,8 @@ fn update(conn: &mut PgConnection) -> QueryResult<()> {
         .execute(conn)?;
     info!("Finished freezing old version_downloads");
 
-    no_arg_sql_function!(refresh_recent_crate_downloads, ());
-    select(refresh_recent_crate_downloads).execute(conn)?;
+    sql_function!(fn refresh_recent_crate_downloads());
+    select(refresh_recent_crate_downloads()).execute(conn)?;
     println!("Finished running refresh_recent_crate_downloads");
 
     Ok(())