More work on converting publish to rust

publish/Cargo.toml

@@ -26,7 +26,7 @@ serde_json.workspace = true
 shlex.workspace = true
 silver-platter.workspace = true
 tokio = { workspace = true, features = ["full"] }
-url.workspace = true
+url = { workspace = true, features = ["serde"] }
 sqlx = { workspace = true, features = ["chrono"] }
 maplit.workspace = true
 prometheus = "0.13.4"

publish/src/bin/janitor-publish.rs

@@ -5,7 +5,7 @@ use janitor_publish::rate_limiter::{
 use std::collections::HashMap;
 use std::net::SocketAddr;
 use std::path::PathBuf;
-use std::sync::{Arc, Mutex};
+use std::sync::{Arc, Mutex, RwLock};
 use url::Url;
 
 #[derive(Parser)]
@@ -62,6 +62,10 @@ struct Args {
     #[clap(long)]
     modify_mp_limit: Option<i32>,
 
+    /// Maximum number of unexpected errors to encounter before stopping.
+    #[clap(long)]
+    unexpected_mp_limit: Option<i32>,
+
     /// External URL
     #[clap(long)]
     external_url: Option<Url>,
@@ -104,10 +108,9 @@ async fn main() -> Result<(), i32> {
             Box::new(NonRateLimiter) as Box<dyn RateLimiter>
         });
 
-    let forge_rate_limiter = Mutex::new(HashMap::new());
+    let forge_rate_limiter = std::sync::Arc::new(RwLock::new(HashMap::new()));
 
-    let vcs_managers = Box::new(janitor::vcs::get_vcs_managers_from_config(config));
-    let vcs_managers: &'static _ = Box::leak(vcs_managers);
+    let vcs_managers = janitor::vcs::get_vcs_managers_from_config(config);
     let db = janitor::state::create_pool(config).await.map_err(|e| {
         log::error!("Failed to create database pool: {}", e);
         1
@@ -136,38 +139,37 @@ async fn main() -> Result<(), i32> {
         .as_deref()
         .map(|redis_location| rslock::LockManager::new(vec![redis_location]));
 
-    let publish_worker = Arc::new(Mutex::new(
-        janitor_publish::PublishWorker::new(
-            args.template_env_path,
-            args.external_url,
-            args.differ_url,
-            redis_async_connection.clone(),
-            lock_manager,
-        )
-        .await,
-    ));
+    let publish_worker = janitor_publish::PublishWorker::new(
+        args.template_env_path,
+        args.external_url,
+        args.differ_url,
+        redis_async_connection.clone(),
+        lock_manager,
+    )
+    .await;
 
     let state = Arc::new(janitor_publish::AppState {
         conn: db.clone(),
         bucket_rate_limiter,
         forge_rate_limiter,
         push_limit: args.push_limit,
+        config,
+        redis: redis_async_connection,
+        vcs_managers,
+        publish_worker,
+        modify_mp_limit: args.modify_mp_limit,
+        unexpected_mp_limit: args.unexpected_mp_limit,
+        gpg: breezyshim::gpg::GPGContext::new(),
+        require_binary_diff: args.require_binary_diff,
     });
 
     if args.once {
-        janitor_publish::publish_pending_ready(
-            state,
-            redis_async_connection.clone(),
-            config,
-            publish_worker.clone(),
-            vcs_managers,
-            args.require_binary_diff,
-        )
-        .await
-        .map_err(|e| {
-            log::error!("Failed to publish pending proposals: {}", e);
-            1
-        })?;
+        janitor_publish::publish_pending_ready(state)
+            .await
+            .map_err(|e| {
+                log::error!("Failed to publish pending proposals: {}", e);
+                1
+            })?;
 
         if let Some(prometheus) = args.prometheus.as_ref() {
             janitor::prometheus::push_to_gateway(
@@ -182,36 +184,15 @@ async fn main() -> Result<(), i32> {
     } else {
         tokio::spawn(janitor_publish::process_queue_loop(
             state.clone(),
-            redis_async_connection.clone(),
-            config,
-            publish_worker.clone(),
-            vcs_managers,
             chrono::Duration::seconds(args.interval),
             !args.no_auto_publish,
-            args.modify_mp_limit,
-            args.require_binary_diff,
         ));
 
         tokio::spawn(janitor_publish::refresh_bucket_mp_counts(state.clone()));
 
-        tokio::spawn(janitor_publish::listen_to_runner(
-            state.clone(),
-            redis_async_connection.clone(),
-            config,
-            publish_worker.clone(),
-            vcs_managers,
-            args.require_binary_diff,
-        ));
+        tokio::spawn(janitor_publish::listen_to_runner(state.clone()));
 
-        let app = janitor_publish::web::app(
-            state.clone(),
-            publish_worker.clone(),
-            vcs_managers,
-            args.require_binary_diff,
-            args.modify_mp_limit,
-            redis_async_connection.clone(),
-            config,
-        );
+        let app = janitor_publish::web::app(state.clone());
 
         // run it
         let addr = SocketAddr::new(args.listen_address, args.port);