Fmt

(This was fixed via `cargo clippy --fix`)

.github/workflows/triage_incoming.yml

@@ -0,0 +1,28 @@
+name: Move new issues into the issue triage board
+
+on:
+  issues:
+    types: [ opened ]
+
+jobs:
+  add_new_issues:
+    name: Add new issues to the triage board
+    runs-on: ubuntu-latest
+    steps:
+      - uses: octokit/graphql-action@v2.x
+        id: add_to_project
+        with:
+          headers: '{"GraphQL-Features": "projects_next_graphql"}'
+          query: |
+              mutation add_to_project($projectid:ID!,$contentid:ID!) {
+                addProjectV2ItemById(input: {projectId: $projectid contentId: $contentid}) {
+                item {
+                  id
+                  }
+                }
+              }
+          projectid: ${{ env.PROJECT_ID }}
+          contentid: ${{ github.event.issue.node_id }}
+        env:
+          PROJECT_ID: "PVT_kwDOAIB0Bs4AFDdZ"
+          GITHUB_TOKEN: ${{ secrets.ELEMENT_BOT_TOKEN }}

.github/workflows/triage_labelled.yml

@@ -0,0 +1,44 @@
+name: Move labelled issues to correct projects
+
+on:
+  issues:
+    types: [ labeled ]
+
+jobs:
+  move_needs_info:
+    name: Move X-Needs-Info on the triage board
+    runs-on: ubuntu-latest
+    if: >
+      contains(github.event.issue.labels.*.name, 'X-Needs-Info')
+    steps:
+      - uses: actions/add-to-project@main
+        id: add_project
+        with:
+          project-url: "https://github.com/orgs/matrix-org/projects/67"
+          github-token: ${{ secrets.ELEMENT_BOT_TOKEN }}
+      - name: Set status
+        env:
+          GITHUB_TOKEN: ${{ secrets.ELEMENT_BOT_TOKEN }}
+        run: |
+          gh api graphql -f query='
+          mutation(
+              $project: ID!
+              $item: ID!
+              $fieldid: ID!
+              $columnid: String!
+            )  {
+            updateProjectV2ItemFieldValue(
+              input: {
+               projectId: $project
+                itemId: $item
+                fieldId: $fieldid
+                value: { 
+                  singleSelectOptionId: $columnid
+                  }
+              }
+            ) {
+              projectV2Item {
+                id
+              }
+            }
+          }' -f project="PVT_kwDOAIB0Bs4AFDdZ" -f item=${{ steps.add_project.outputs.itemId }} -f fieldid="PVTSSF_lADOAIB0Bs4AFDdZzgC6ZA4" -f columnid=ba22e43c --silent

.vscode/settings.json

@@ -1,3 +0,0 @@
-{
-    "rust-analyzer.checkOnSave.command": "clippy"
-}

CONTRIBUTING.md

@@ -0,0 +1,80 @@
+# Contributing
+
+## Sign off
+
+In order to have a concrete record that your contribution is intentional
+and you agree to license it under the same terms as the project's license, we've adopted the
+same lightweight approach that the Linux Kernel
+[submitting patches process](
+https://www.kernel.org/doc/html/latest/process/submitting-patches.html#sign-your-work-the-developer-s-certificate-of-origin>),
+[Docker](https://github.com/docker/docker/blob/master/CONTRIBUTING.md), and many other
+projects use: the DCO ([Developer Certificate of Origin](http://developercertificate.org/)).
+This is a simple declaration that you wrote
+the contribution or otherwise have the right to contribute it to Matrix:
+
+```
+Developer Certificate of Origin
+Version 1.1
+
+Copyright (C) 2004, 2006 The Linux Foundation and its contributors.
+660 York Street, Suite 102,
+San Francisco, CA 94110 USA
+
+Everyone is permitted to copy and distribute verbatim copies of this
+license document, but changing it is not allowed.
+
+Developer's Certificate of Origin 1.1
+
+By making a contribution to this project, I certify that:
+
+(a) The contribution was created in whole or in part by me and I
+    have the right to submit it under the open source license
+    indicated in the file; or
+
+(b) The contribution is based upon previous work that, to the best
+    of my knowledge, is covered under an appropriate open source
+    license and I have the right under that license to submit that
+    work with modifications, whether created in whole or in part
+    by me, under the same open source license (unless I am
+    permitted to submit under a different license), as indicated
+    in the file; or
+
+(c) The contribution was provided directly to me by some other
+    person who certified (a), (b) or (c) and I have not modified
+    it.
+
+(d) I understand and agree that this project and the contribution
+    are public and that a record of the contribution (including all
+    personal information I submit with it, including my sign-off) is
+    maintained indefinitely and may be redistributed consistent with
+    this project or the open source license(s) involved.
+```
+
+If you agree to this for your contribution, then all that's needed is to
+include the line in your commit or pull request comment:
+
+```
+Signed-off-by: Your Name <your@email.example.org>
+```
+
+We accept contributions under a legally identifiable name, such as
+your name on government documentation or common-law names (names
+claimed by legitimate usage or repute). Unfortunately, we cannot
+accept anonymous contributions at this time.
+
+Git allows you to add this signoff automatically when using the `-s`
+flag to `git commit`, which uses the name and email set in your
+`user.name` and `user.email` git configs.
+
+### Private Sign off
+
+If you would like to provide your legal name privately to the Matrix.org
+Foundation (instead of in a public commit or comment), you can do so
+by emailing your legal name and a link to the pull request to
+[dco@matrix.org](mailto:dco@matrix.org?subject=Private%20sign%20off).
+It helps to include "sign off" or similar in the subject line. You will then
+be instructed further.
+
+Once private sign off is complete, doing so for future contributions will not
+be required.
+

Cargo.toml

@@ -9,8 +9,7 @@ version = "0.1.0"
 edition = "2018"
 
 [dependencies]
-clap = "2.33.0"
+indicatif = "0.17.0"
-indicatif = "0.16.0"
 openssl = "0.10.32"
 postgres = "0.19.0"
 postgres-openssl = "0.5.0"
@@ -19,7 +18,7 @@ rayon = "1.3.0"
 string_cache = "0.8.0"
 env_logger = "0.9.0"
 log = "0.4.14"
-pyo3-log = "0.4.0"
+pyo3-log = "0.7.0"
 log-panics = "2.0.0"
 
 [dependencies.state-map]
@@ -29,12 +28,16 @@ git = "https://github.com/matrix-org/rust-matrix-state-map"
 [lib]
 crate-type = ["cdylib", "rlib"]
 
+[dependencies.clap]
+version = "4.0.15"
+features = ["cargo"]
+
 [dependencies.pyo3]
-version = "0.14.1"
+version = "0.17.1"
-features = ["extension-module","abi3-py36"]
+features = ["extension-module"]
 
 [dependencies.tikv-jemallocator]
-version = "0.4.1"
+version = "0.5.0"
 optional = true
 
 [features]

Dockerfile

@@ -1,4 +1,4 @@
-FROM rust:alpine AS builder
+FROM docker.io/rust:alpine AS builder
 
 RUN apk add python3 musl-dev pkgconfig openssl-dev make
 
@@ -14,7 +14,7 @@ WORKDIR /opt/synapse-compressor/synapse_auto_compressor/
 
 RUN cargo build
 
-FROM alpine
+FROM docker.io/alpine
 
 RUN apk add --no-cache libgcc
 

README.md

@@ -26,6 +26,9 @@ periodically.
 This tool requires `cargo` to be installed. See https://www.rust-lang.org/tools/install
 for instructions on how to do this.
 
+This project follows the deprecation policy of [Synapse](https://matrix-org.github.io/synapse/latest/deprecation_policy.html)
+on Rust and will assume a recent stable version of Rust and the ability to fetch a more recent one if necessary.
+
 To build `synapse_auto_compressor`, clone this repository and navigate to the
 `synapse_auto_compressor/` subdirectory. Then execute `cargo build`.
 

compressor_integration_tests/Cargo.toml

@@ -7,7 +7,7 @@ edition = "2018"
 
 [dependencies]
 string_cache = "0.8.0"
-serial_test = "0.5.1"
+serial_test = "0.9.0"
 openssl = "0.10.32"
 postgres = "0.19.0"
 postgres-openssl = "0.5.0"

compressor_integration_tests/src/lib.rs

@@ -4,7 +4,12 @@ use postgres::{fallible_iterator::FallibleIterator, Client};
 use postgres_openssl::MakeTlsConnector;
 use rand::{distributions::Alphanumeric, thread_rng, Rng};
 use state_map::StateMap;
-use std::{borrow::Cow, collections::BTreeMap, env, fmt};
+use std::{
+    borrow::Cow,
+    collections::BTreeMap,
+    env,
+    fmt::{self, Write as _},
+};
 use string_cache::DefaultAtom as Atom;
 
 use synapse_compress_state::StateGroupEntry;
@@ -23,47 +28,48 @@ pub fn add_contents_to_database(room_id: &str, state_group_map: &BTreeMap<i64, S
     let mut client = Client::connect(DB_URL, connector).unwrap();
 
     // build up the query
-    let mut sql = "".to_string();
+    let mut sql = String::new();
+
+    let room_id = PGEscape(room_id);
+    let event_id = PGEscape("left_blank");
 
     for (sg, entry) in state_group_map {
         // create the entry for state_groups
-        sql.push_str(&format!(
+        writeln!(
-            "INSERT INTO state_groups (id, room_id, event_id) VALUES ({},{},{});\n",
+            sql,
-            sg,
+            "INSERT INTO state_groups (id, room_id, event_id) \
-            PGEscape(room_id),
+             VALUES ({sg}, {room_id}, {event_id});",
-            PGEscape("left_blank")
+        )
-        ));
+        .expect("Writing to a String cannot fail");
 
         // create the entry in state_group_edges IF exists
         if let Some(prev_sg) = entry.prev_state_group {
-            sql.push_str(&format!(
+            writeln!(
-                "INSERT INTO state_group_edges (state_group, prev_state_group) VALUES ({}, {});\n",
+                sql,
-                sg, prev_sg
+                "INSERT INTO state_group_edges (state_group, prev_state_group) \
-            ));
+                 VALUES ({sg}, {prev_sg});",
+            )
+            .unwrap();
         }
 
         // write entry for each row in delta
         if !entry.state_map.is_empty() {
-            sql.push_str("INSERT INTO state_groups_state (state_group, room_id, type, state_key, event_id) VALUES");
+            sql.push_str(
+                "INSERT INTO state_groups_state \
+                 (state_group, room_id, type, state_key, event_id) \
+                 VALUES\n",
+            );
 
-            let mut first = true;
             for ((t, s), e) in entry.state_map.iter() {
-                if first {
+                let t = PGEscape(t);
-                    sql.push_str("     ");
+                let s = PGEscape(s);
-                    first = false;
+                let e = PGEscape(e);
-                } else {
+
-                    sql.push_str("    ,");
+                writeln!(sql, "    ({sg}, {room_id}, {t}, {s}, {e}),").unwrap();
-                }
-                sql.push_str(&format!(
-                    "({}, {}, {}, {}, {})",
-                    sg,
-                    PGEscape(room_id),
-                    PGEscape(t),
-                    PGEscape(s),
-                    PGEscape(e)
-                ));
             }
-            sql.push_str(";\n");
+
+            // Replace the last comma with a semicolon
+            sql.replace_range((sql.len() - 2).., ";\n");
         }
     }
 

src/compressor.rs

@@ -30,7 +30,7 @@
 
 use indicatif::{ProgressBar, ProgressStyle};
 use state_map::StateMap;
-use std::collections::BTreeMap;
+use std::{collections::BTreeMap, time::Duration};
 use string_cache::DefaultAtom as Atom;
 
 use super::{collapse_state_maps, StateGroupEntry};
@@ -156,7 +156,7 @@ impl<'a> Compressor<'a> {
     ) -> Compressor<'a> {
         let levels = level_info
             .iter()
-            .map(|l| Level::restore((*l).max_length, (*l).current_chain_length, (*l).head))
+            .map(|l| Level::restore(l.max_length, l.current_chain_length, l.head))
             .collect();
 
         let mut compressor = Compressor {
@@ -181,16 +181,18 @@ impl<'a> Compressor<'a> {
             panic!("Can only call `create_new_tree` once");
         }
 
-        let pb: ProgressBar = if cfg!(feature = "no-progress-bars") {
+        let pb = if cfg!(feature = "no-progress-bars") {
             ProgressBar::hidden()
         } else {
             ProgressBar::new(self.original_state_map.len() as u64)
         };
         pb.set_style(
-            ProgressStyle::default_bar().template("[{elapsed_precise}] {bar} {pos}/{len} {msg}"),
+            ProgressStyle::default_bar()
+                .template("[{elapsed_precise}] {bar} {pos}/{len} {msg}")
+                .unwrap(),
         );
         pb.set_message("state groups");
-        pb.enable_steady_tick(100);
+        pb.enable_steady_tick(Duration::from_millis(100));
 
         for (&state_group, entry) in self.original_state_map {
             // Check whether this entry is in_range or is just present in the map due to being

src/compressor/compressor_tests.rs

@@ -93,7 +93,10 @@ fn create_new_tree_does_nothing_if_already_compressed() {
 
     for i in 0i64..=13i64 {
         // edge from map
-        let prev = initial_edges.get(&i).copied();
+        let pred_group = initial_edges.get(&i);
+
+        // Need Option<i64> not Option<&i64>
+        let prev = pred_group.copied();
 
         // insert that edge into the initial map
         initial.insert(

src/compressor/stats_tests.rs

@@ -142,7 +142,10 @@ fn stats_correct_if_no_changes() {
 
     for i in 0i64..=13i64 {
         // edge from map
-        let prev = initial_edges.get(&i).copied();
+        let pred_group = initial_edges.get(&i);
+
+        // Need Option<i64> not Option<&i64>
+        let prev = pred_group.copied();
 
         // insert that edge into the initial map
         initial.insert(

src/database.rs

@@ -18,7 +18,7 @@ use openssl::ssl::{SslConnector, SslMethod, SslVerifyMode};
 use postgres::{fallible_iterator::FallibleIterator, types::ToSql, Client};
 use postgres_openssl::MakeTlsConnector;
 use rand::{distributions::Alphanumeric, thread_rng, Rng};
-use std::{borrow::Cow, collections::BTreeMap, fmt};
+use std::{borrow::Cow, collections::BTreeMap, fmt, time::Duration};
 
 use crate::{compressor::Level, generate_sql};
 
@@ -237,15 +237,9 @@ fn load_map_from_db(
         let mut missing_sgs: Vec<_> = state_group_map
             .iter()
             .filter_map(|(_sg, entry)| {
-                if let Some(prev_sg) = entry.prev_state_group {
+                entry
-                    if state_group_map.contains_key(&prev_sg) {
+                    .prev_state_group
-                        None
+                    .filter(|&prev_sg| !state_group_map.contains_key(&prev_sg))
-                    } else {
-                        Some(prev_sg)
-                    }
-                } else {
-                    None
-                }
             })
             .collect();
 
@@ -372,15 +366,17 @@ fn get_initial_data_from_db(
     // Copy the data from the database into a map
     let mut state_group_map: BTreeMap<i64, StateGroupEntry> = BTreeMap::new();
 
-    let pb: ProgressBar = if cfg!(feature = "no-progress-bars") {
+    let pb = if cfg!(feature = "no-progress-bars") {
         ProgressBar::hidden()
     } else {
         ProgressBar::new_spinner()
     };
     pb.set_style(
-        ProgressStyle::default_spinner().template("{spinner} [{elapsed}] {pos} rows retrieved"),
+        ProgressStyle::default_spinner()
+            .template("{spinner} [{elapsed}] {pos} rows retrieved")
+            .unwrap(),
     );
-    pb.enable_steady_tick(100);
+    pb.enable_steady_tick(Duration::from_millis(100));
 
     while let Some(row) = rows.next().unwrap() {
         // The row in the map to copy the data to
@@ -536,16 +532,18 @@ pub fn send_changes_to_db(
     debug!("Writing changes...");
 
     // setup the progress bar
-    let pb: ProgressBar = if cfg!(feature = "no-progress-bars") {
+    let pb = if cfg!(feature = "no-progress-bars") {
         ProgressBar::hidden()
     } else {
         ProgressBar::new(old_map.len() as u64)
     };
     pb.set_style(
-        ProgressStyle::default_bar().template("[{elapsed_precise}] {bar} {pos}/{len} {msg}"),
+        ProgressStyle::default_bar()
+            .template("[{elapsed_precise}] {bar} {pos}/{len} {msg}")
+            .unwrap(),
     );
     pb.set_message("state groups");
-    pb.enable_steady_tick(100);
+    pb.enable_steady_tick(Duration::from_millis(100));
 
     for sql_transaction in generate_sql(old_map, new_map, room_id) {
         if sql_transaction.is_empty() {

src/lib.rs

@@ -23,11 +23,14 @@
 use log::{info, warn, LevelFilter};
 use pyo3::{exceptions, prelude::*};
 
-use clap::{crate_authors, crate_description, crate_name, crate_version, value_t, App, Arg};
+use clap::{crate_authors, crate_description, crate_name, crate_version, Arg, Command};
 use indicatif::{ProgressBar, ProgressStyle};
 use rayon::prelude::*;
 use state_map::StateMap;
-use std::{collections::BTreeMap, convert::TryInto, fs::File, io::Write, str::FromStr};
+use std::{
+    collections::BTreeMap, convert::TryInto, fmt::Write as _, fs::File, io::Write, str::FromStr,
+    time::Duration,
+};
 use string_cache::DefaultAtom as Atom;
 
 mod compressor;
@@ -49,7 +52,7 @@ pub struct StateGroupEntry {
 }
 
 /// Helper struct for parsing the `level_sizes` argument.
-#[derive(PartialEq, Debug)]
+#[derive(PartialEq, Debug, Clone)]
 struct LevelSizes(Vec<usize>);
 
 impl FromStr for LevelSizes {
@@ -117,13 +120,13 @@ pub struct Config {
 impl Config {
     /// Build up config from command line arguments
     pub fn parse_arguments() -> Config {
-        let matches = App::new(crate_name!())
+        let matches = Command::new(crate_name!())
         .version(crate_version!())
         .author(crate_authors!("\n"))
         .about(crate_description!())
         .arg(
-            Arg::with_name("postgres-url")
+            Arg::new("postgres-url")
-                .short("p")
+                .short('p')
                 .value_name("POSTGRES_LOCATION")
                 .help("The configruation for connecting to the postgres database.")
                 .long_help(concat!(
@@ -133,64 +136,69 @@ impl Config {
                     "See https://docs.rs/tokio-postgres/0.7.2/tokio_postgres/config/struct.Config.html ",
                     "for the full details."
                 ))
-                .takes_value(true)
+                .num_args(1)
                 .required(true),
         ).arg(
-            Arg::with_name("room_id")
+            Arg::new("room_id")
-                .short("r")
+                .short('r')
                 .value_name("ROOM_ID")
                 .help("The room to process")
                 .long_help(concat!(
                     "The room to process. This is the value found in the rooms table of the database",
                     " not the common name for the room - is should look like: \"!wOlkWNmgkAZFxbTaqj:matrix.org\""
                 ))
-                .takes_value(true)
+                .num_args(1)
                 .required(true),
         ).arg(
-            Arg::with_name("min_state_group")
+            Arg::new("min_state_group")
-                .short("b")
+                .short('b')
                 .value_name("MIN_STATE_GROUP")
+                .value_parser(clap::value_parser!(i64))
                 .help("The state group to start processing from (non inclusive)")
-                .takes_value(true)
+                .num_args(1)
                 .required(false),
         ).arg(
-            Arg::with_name("min_saved_rows")
+            Arg::new("min_saved_rows")
-                .short("m")
+                .short('m')
                 .value_name("COUNT")
+                .value_parser(clap::value_parser!(i32))
                 .help("Abort if fewer than COUNT rows would be saved")
                 .long_help("If the compressor cannot save this many rows from the database then it will stop early")
-                .takes_value(true)
+                .num_args(1)
                 .required(false),
         ).arg(
-            Arg::with_name("groups_to_compress")
+            Arg::new("groups_to_compress")
-                .short("n")
+                .short('n')
                 .value_name("GROUPS_TO_COMPRESS")
+                .value_parser(clap::value_parser!(i64))
                 .help("How many groups to load into memory to compress")
                 .long_help(concat!(
                     "How many groups to load into memory to compress (starting from",
                     " the 1st group in the room or the group specified by -s)"))
-                .takes_value(true)
+                .num_args(1)
                 .required(false),
         ).arg(
-            Arg::with_name("output_file")
+            Arg::new("output_file")
-                .short("o")
+                .short('o')
                 .value_name("FILE")
                 .help("File to output the changes to in SQL")
-                .takes_value(true),
+                .num_args(1),
         ).arg(
-            Arg::with_name("max_state_group")
+            Arg::new("max_state_group")
-                .short("s")
+                .short('s')
                 .value_name("MAX_STATE_GROUP")
+                .value_parser(clap::value_parser!(i64))
                 .help("The maximum state group to process up to")
                 .long_help(concat!(
                     "If a max_state_group is specified then only state groups with id's lower",
                     " than this number are able to be compressed."))
-                .takes_value(true)
+                .num_args(1)
                 .required(false),
         ).arg(
-            Arg::with_name("level_sizes")
+            Arg::new("level_sizes")
-                .short("l")
+                .short('l')
                 .value_name("LEVELS")
+                .value_parser(clap::value_parser!(LevelSizes))
                 .help("Sizes of each new level in the compression algorithm, as a comma separated list.")
                 .long_help(concat!(
                     "Sizes of each new level in the compression algorithm, as a comma separated list.",
@@ -203,32 +211,36 @@ impl Config {
                     " iterations needed to fetch a given set of state.",
                 ))
                 .default_value("100,50,25")
-                .takes_value(true),
+                .num_args(1),
         ).arg(
-            Arg::with_name("transactions")
+            Arg::new("transactions")
-                .short("t")
+                .short('t')
+                .action(clap::ArgAction::SetTrue)
                 .help("Whether to wrap each state group change in a transaction")
                 .long_help(concat!("If this flag is set then then each change to a particular",
                     " state group is wrapped in a transaction. This should be done if you wish to",
                     " apply the changes while synapse is still running."))
                 .requires("output_file"),
         ).arg(
-            Arg::with_name("graphs")
+            Arg::new("graphs")
-                .short("g")
+                .short('g')
+                .action(clap::ArgAction::SetTrue)
                 .help("Output before and after graphs")
                 .long_help(concat!("If this flag is set then output the node and edge information for",
                     " the state_group directed graph built up from the predecessor state_group links.",
                     " These can be looked at in something like Gephi (https://gephi.org)")),
         ).arg(
-            Arg::with_name("commit_changes")
+            Arg::new("commit_changes")
-                .short("c")
+                .short('c')
+                .action(clap::ArgAction::SetTrue)
                 .help("Commit changes to the database")
                 .long_help(concat!("If this flag is set then the changes the compressor makes will",
                     " be committed to the database. This should be safe to use while synapse is running",
                     " as it assumes by default that the transactions flag is set")),
         ).arg(
-            Arg::with_name("no_verify")
+            Arg::new("no_verify")
-                .short("N")
+                .short('N')
+                .action(clap::ArgAction::SetTrue)
                 .help("Do not double-check that the compression was performed correctly")
                 .long_help(concat!("If this flag is set then the verification of the compressed",
                     " state groups, which compares them to the original groups, is skipped. This",
@@ -236,43 +248,27 @@ impl Config {
         ).get_matches();
 
         let db_url = matches
-            .value_of("postgres-url")
+            .get_one::<String>("postgres-url")
             .expect("db url should be required");
 
-        let output_file = matches.value_of("output_file").map(|path| {
+        let output_file = matches.get_one::<String>("output_file").map(|path| {
             File::create(path).unwrap_or_else(|e| panic!("Unable to create output file: {}", e))
         });
 
         let room_id = matches
-            .value_of("room_id")
+            .get_one::<String>("room_id")
             .expect("room_id should be required since no file");
 
-        let min_state_group = matches
+        let min_state_group = matches.get_one("min_state_group").copied();
-            .value_of("min_state_group")
+        let groups_to_compress = matches.get_one("groups_to_compress").copied();
-            .map(|s| s.parse().expect("min_state_group must be an integer"));
+        let min_saved_rows = matches.get_one("min_saved_rows").copied();
+        let max_state_group = matches.get_one("max_state_group").copied();
+        let level_sizes = matches.get_one("level_sizes").cloned().unwrap();
 
-        let groups_to_compress = matches
+        let transactions = matches.get_flag("transactions");
-            .value_of("groups_to_compress")
+        let graphs = matches.get_flag("graphs");
-            .map(|s| s.parse().expect("groups_to_compress must be an integer"));
+        let commit_changes = matches.get_flag("commit_changes");
-
+        let verify = !matches.get_flag("no_verify");
-        let min_saved_rows = matches
-            .value_of("min_saved_rows")
-            .map(|v| v.parse().expect("COUNT must be an integer"));
-
-        let max_state_group = matches
-            .value_of("max_state_group")
-            .map(|s| s.parse().expect("max_state_group must be an integer"));
-
-        let level_sizes = value_t!(matches, "level_sizes", LevelSizes)
-            .unwrap_or_else(|e| panic!("Unable to parse level_sizes: {}", e));
-
-        let transactions = matches.is_present("transactions");
-
-        let graphs = matches.is_present("graphs");
-
-        let commit_changes = matches.is_present("commit_changes");
-
-        let verify = !matches.is_present("no_verify");
 
         Config {
             db_url: String::from(db_url),
@@ -423,8 +419,7 @@ fn generate_sql<'a>(
     new_map: &'a BTreeMap<i64, StateGroupEntry>,
     room_id: &'a str,
 ) -> impl Iterator<Item = String> + 'a {
-    old_map.iter().map(move |(sg,old_entry)| {
+    old_map.iter().map(move |(sg, old_entry)| {
-
         let new_entry = &new_map[sg];
 
         // Check if the new map has a different entry for this state group
@@ -434,48 +429,50 @@ fn generate_sql<'a>(
             let mut sql = String::new();
 
             // remove the current edge
-            sql.push_str(&format!(
+            writeln!(
-                "DELETE FROM state_group_edges WHERE state_group = {};\n",
+                sql,
-                sg
+                "DELETE FROM state_group_edges WHERE state_group = {sg};",
-            ));
+            )
+            .expect("Writing to a String cannot fail");
 
             // if the new entry has a predecessor then put that into state_group_edges
             if let Some(prev_sg) = new_entry.prev_state_group {
-                sql.push_str(&format!("INSERT INTO state_group_edges (state_group, prev_state_group) VALUES ({}, {});\n", sg, prev_sg));
+                writeln!(
+                    sql,
+                    "INSERT INTO state_group_edges (state_group, prev_state_group) \
+                     VALUES ({sg}, {prev_sg});",
+                )
+                .unwrap();
             }
 
             // remove the current deltas for this state group
-            sql.push_str(&format!(
+            writeln!(
-                "DELETE FROM state_groups_state WHERE state_group = {};\n",
+                sql,
-                sg
+                "DELETE FROM state_groups_state WHERE state_group = {sg};",
-            ));
+            )
+            .unwrap();
 
             if !new_entry.state_map.is_empty() {
                 // place all the deltas for the state group in the new map into state_groups_state
-                sql.push_str("INSERT INTO state_groups_state (state_group, room_id, type, state_key, event_id) VALUES\n");
+                sql.push_str(
+                    "INSERT INTO state_groups_state \
+                     (state_group, room_id, type, state_key, event_id) \
+                     VALUES\n",
+                );
 
-                let mut first = true;
+                let room_id = PGEscape(room_id);
                 for ((t, s), e) in new_entry.state_map.iter() {
-                    // Add a comma at the start if not the first row to be inserted
+                    let t = PGEscape(t);
-                    if first {
+                    let s = PGEscape(s);
-                        sql.push_str("     ");
+                    let e = PGEscape(e);
-                        first = false;
-                    } else {
-                        sql.push_str("    ,");
-                    }
 
-                    // write the row to be insterted of the form:
+                    // write the row to be inserted of the form:
                     // (state_group, room_id, type, state_key, event_id)
-                    sql.push_str(&format!(
+                    writeln!(sql, "    ({sg}, {room_id}, {t}, {s}, {e}),").unwrap();
-                        "({}, {}, {}, {}, {})",
-                        sg,
-                        PGEscape(room_id),
-                        PGEscape(t),
-                        PGEscape(s),
-                        PGEscape(e)
-                    ));
                 }
-                sql.push_str(";\n");
+
+                // Replace the last comma with a semicolon
+                sql.replace_range((sql.len() - 2).., ";\n");
             }
 
             sql
@@ -507,16 +504,18 @@ fn output_sql(
 
     info!("Writing changes...");
 
-    let pb: ProgressBar = if cfg!(feature = "no-progress-bars") {
+    let pb = if cfg!(feature = "no-progress-bars") {
         ProgressBar::hidden()
     } else {
         ProgressBar::new(old_map.len() as u64)
     };
     pb.set_style(
-        ProgressStyle::default_bar().template("[{elapsed_precise}] {bar} {pos}/{len} {msg}"),
+        ProgressStyle::default_bar()
+            .template("[{elapsed_precise}] {bar} {pos}/{len} {msg}")
+            .unwrap(),
     );
     pb.set_message("state groups");
-    pb.enable_steady_tick(100);
+    pb.enable_steady_tick(Duration::from_millis(100));
 
     if let Some(output) = &mut config.output_file {
         for mut sql_transaction in generate_sql(old_map, new_map, &config.room_id) {
@@ -564,7 +563,7 @@ pub fn continue_run(
     let (state_group_map, max_group_found) =
         database::reload_data_from_db(db_url, room_id, start, Some(chunk_size), level_info)?;
 
-    let original_num_rows = state_group_map.iter().map(|(_, v)| v.state_map.len()).sum();
+    let original_num_rows = state_group_map.values().map(|v| v.state_map.len()).sum();
 
     // Now we actually call the compression algorithm.
     let compressor = Compressor::compress_from_save(&state_group_map, level_info);
@@ -621,16 +620,18 @@ fn check_that_maps_match(
 ) {
     info!("Checking that state maps match...");
 
-    let pb: ProgressBar = if cfg!(feature = "no-progress-bars") {
+    let pb = if cfg!(feature = "no-progress-bars") {
         ProgressBar::hidden()
     } else {
         ProgressBar::new(old_map.len() as u64)
     };
     pb.set_style(
-        ProgressStyle::default_bar().template("[{elapsed_precise}] {bar} {pos}/{len} {msg}"),
+        ProgressStyle::default_bar()
+            .template("[{elapsed_precise}] {bar} {pos}/{len} {msg}")
+            .unwrap(),
     );
     pb.set_message("state groups");
-    pb.enable_steady_tick(100);
+    pb.enable_steady_tick(Duration::from_millis(100));
 
     // Now let's iterate through and assert that the state for each group
     // matches between the two versions.

synapse_auto_compressor/Cargo.toml

@@ -1,11 +1,11 @@
 [package]
 name = "synapse_auto_compressor"
 authors = ["William Ashton"]
-version = "0.1.2"
+version = "0.1.3"
 edition = "2018"
 
 [package.metadata.maturin]
-requires-python = ">=3.6"
+requires-python = ">=3.7"
 project-url = {Source = "https://github.com/matrix-org/rust-synapse-compress-state"}
 classifier = [
     "Development Status :: 4 - Beta",
@@ -13,24 +13,34 @@ classifier = [
 ]
 
 [dependencies]
-clap = "2.33.0"
 openssl = "0.10.32"
 postgres = "0.19.0"
 postgres-openssl = "0.5.0"
-tikv-jemallocator = "0.4.1"
 rand = "0.8.0"
-serial_test = "0.5.1"
+serial_test = "0.9.0"
 synapse_compress_state = { path = "../", features = ["no-progress-bars"] }
 env_logger = "0.9.0"
 log = "0.4.14"
 log-panics = "2.0.0"
 anyhow = "1.0.42"
-pyo3-log = "0.4.0"
+pyo3-log = "0.7.0"
 
 # Needed for pyo3 support
 [lib]
 crate-type = ["cdylib", "rlib"]
 
+[dependencies.clap]
+version = "4.0.15"
+features = ["cargo"]
+
 [dependencies.pyo3]
-version = "0.14.1"
+version = "0.17.1"
-features = ["extension-module","abi3-py36"]
+features = ["extension-module"]
+
+[dependencies.tikv-jemallocator]
+version = "0.5.0"
+optional = true
+
+[features]
+default = ["jemalloc"]
+jemalloc = ["tikv-jemallocator"]

synapse_auto_compressor/src/lib.rs

@@ -26,7 +26,7 @@ pub mod state_saving;
 ///
 /// This is needed since FromStr cannot be implemented for structs
 /// that aren't defined in this scope
-#[derive(PartialEq, Debug)]
+#[derive(PartialEq, Eq, Debug, Clone)]
 pub struct LevelInfo(pub Vec<Level>);
 
 // Implement FromStr so that an argument of the form "100,50,25"

synapse_auto_compressor/src/main.rs

@@ -16,10 +16,11 @@
 //! the state_compressor_state table so that the compressor can seemlesly
 //! continue from where it left off.
 
+#[cfg(feature = "jemalloc")]
 #[global_allocator]
 static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc;
 
-use clap::{crate_authors, crate_description, crate_name, crate_version, value_t, App, Arg};
+use clap::{crate_authors, crate_description, crate_name, crate_version, Arg, Command};
 use log::LevelFilter;
 use std::env;
 use synapse_auto_compressor::{manager, state_saving, LevelInfo};
@@ -50,13 +51,13 @@ fn main() {
     log::info!("synapse_auto_compressor started");
 
     // parse the command line arguments using the clap crate
-    let arguments = App::new(crate_name!())
+    let arguments = Command::new(crate_name!())
         .version(crate_version!())
         .author(crate_authors!("\n"))
         .about(crate_description!())
         .arg(
-            Arg::with_name("postgres-url")
+            Arg::new("postgres-url")
-                .short("p")
+                .short('p')
                 .value_name("POSTGRES_LOCATION")
                 .help("The configruation for connecting to the postgres database.")
                 .long_help(concat!(
@@ -66,12 +67,13 @@ fn main() {
                     "See https://docs.rs/tokio-postgres/0.7.2/tokio_postgres/config/struct.Config.html ",
                     "for the full details."
                 ))
-                .takes_value(true)
+                .num_args(1)
                 .required(true),
         ).arg(
-            Arg::with_name("chunk_size")
+            Arg::new("chunk_size")
-                .short("c")
+                .short('c')
                 .value_name("COUNT")
+                .value_parser(clap::value_parser!(i64))
                 .help("The maximum number of state groups to load into memroy at once")
                 .long_help(concat!(
                     "The number of state_groups to work on at once. All of the entries",
@@ -82,12 +84,13 @@ fn main() {
                     " chunk as a whole (which may well happen in rooms with lots",
                     " of backfill in) then the entire chunk is skipped.)",
                 ))
-                .takes_value(true)
+                .num_args(1)
                 .required(true),
         ).arg(
-            Arg::with_name("default_levels")
+            Arg::new("default_levels")
-                .short("l")
+                .short('l')
                 .value_name("LEVELS")
+                .value_parser(clap::value_parser!(LevelInfo))
                 .help("Sizes of each new level in the compression algorithm, as a comma separated list.")
                 .long_help(concat!(
                     "Sizes of each new level in the compression algorithm, as a comma separated list.",
@@ -100,40 +103,43 @@ fn main() {
                     " iterations needed to fetch a given set of state.",
                 ))
                 .default_value("100,50,25")
-                .takes_value(true)
+                .num_args(1)
                 .required(false),
         ).arg(
-            Arg::with_name("number_of_chunks")
+            Arg::new("number_of_chunks")
-                .short("n")
+                .short('n')
                 .value_name("CHUNKS_TO_COMPRESS")
+                .value_parser(clap::value_parser!(i64))
                 .help("The number of chunks to compress")
                 .long_help(concat!(
                     "This many chunks of the database will be compressed. The higher this number is set to, ",
                     "the longer the compressor will run for."
                 ))
-                .takes_value(true)
+                .num_args(1)
                 .required(true),
         ).get_matches();
 
     // The URL of the database
     let db_url = arguments
-        .value_of("postgres-url")
+        .get_one::<String>("postgres-url")
         .expect("A database url is required");
 
     // The number of state groups to work on at once
     let chunk_size = arguments
-        .value_of("chunk_size")
+        .get_one("chunk_size")
-        .map(|s| s.parse().expect("chunk_size must be an integer"))
+        .copied()
         .expect("A chunk size is required");
 
     // The default structure to use when compressing
-    let default_levels = value_t!(arguments, "default_levels", LevelInfo)
+    let default_levels = arguments
-        .unwrap_or_else(|e| panic!("Unable to parse default levels: {}", e));
+        .get_one::<LevelInfo>("default_levels")
+        .cloned()
+        .unwrap();
 
     // The number of rooms to compress with this tool
     let number_of_chunks = arguments
-        .value_of("number_of_chunks")
+        .get_one("number_of_chunks")
-        .map(|s| s.parse().expect("number_of_chunks must be an integer"))
+        .copied()
         .expect("number_of_chunks is required");
 
     // Connect to the database and create the 2 tables this tool needs