Auto merge of #3423 - pietroalbini:downloads-counter-tests, r=jtgeibel

Adds tests for DownloadsCounter

This fixes #3416 by adding tests for `DownloadsCounter`. There are a few notable things in this PR I'd like to highlight.

First off, the tests are not implemented in `src/tests`, but are implemented in a `#[cfg(test)] mod tests {}` in the same source file. While this resulted in some code duplication I had to do this because testing `persist_next_shard` requires access to the internals of `DownloadsCounter` and I'd prefer not to expose them.

This PR also fixes a bug I discovered while writing the test for incrementing a missing version. Inserting a missing version ID in the `version_downloads` table doesn't work thanks to the foreign key, and aborts the whole `INSERT` statement (losing all the downloads in that shard). This could happen in production if a version is deleted between calling `increment()` and persisting that shard.

To fix that bug I changed the code to do a `SELECT FOR SHARE` from the `versions` table of the version IDs we're about to persist, discarding the versions that are not returned by the query. The `FOR SHARE` clause does the equivalent of `RwLock::read`, preventing other transactions from updating or deleting the selected rows while allowing other `SELECT`s and `SELECT FOR SHARE`s.

There were also some refactorings to make the tests better to write.

This PR is best reviewed commit-by-commit.
r? `@jtgeibel`

Author: bors

src/bin/server.rs

@@ -138,8 +138,9 @@ fn main() -> Result<(), Box<dyn std::error::Error>> {
     rt.block_on(async { server.await.unwrap() });
 
     println!("Persisting remaining downloads counters");
-    if let Err(err) = app.downloads_counter.persist_all_shards(&app) {
-        println!("downloads_counter error: {}", err);
+    match app.downloads_counter.persist_all_shards(&app) {
+        Ok(stats) => stats.log(),
+        Err(err) => println!("downloads_counter error: {}", err),
     }
 
     println!("Server has gracefully shutdown!");
@@ -154,8 +155,9 @@ fn downloads_counter_thread(app: Arc<App>) {
     std::thread::spawn(move || loop {
         std::thread::sleep(interval);
 
-        if let Err(err) = app.downloads_counter.persist_next_shard(&app) {
-            println!("downloads_counter error: {}", err);
+        match app.downloads_counter.persist_next_shard(&app) {
+            Ok(stats) => stats.log(),
+            Err(err) => println!("downloads_counter error: {}", err),
         }
     });
 }

src/db.rs

@@ -130,3 +130,10 @@ impl CustomizeConnection<PgConnection, r2d2::Error> for ConnectionConfig {
         Ok(())
     }
 }
+
+#[cfg(test)]
+pub(crate) fn test_conn() -> PgConnection {
+    let conn = PgConnection::establish(&crate::env("TEST_DATABASE_URL")).unwrap();
+    conn.begin_test_transaction().unwrap();
+    conn
+}