Exclude backfill request types from the general worker

Summary:
**Context:** We want backfills to be handled by a separate set of async request workers, to better allow controlling scaling, execution concurrency etc. We'll use the same framework, the same request queue etc, but have one set of workers handling the backfill requests, and one handling the remaining requests.

**This Diff:** Exclude backfill types from the 'general' async requests workers.

Differential Revision: D100604598

fbshipit-source-id: c7978009bf45c8016d0fead04d998930a163cd65

Author: Julian Priestley

eden/mononoke/features/async_requests/worker/BUCK

@@ -22,6 +22,7 @@ rust_binary(
         "//eden/mononoke/features/async_requests:async_requests",
         "//eden/mononoke/features/async_requests:async_requests_client",
         "//eden/mononoke/features/async_requests:requests_table",
+        "//eden/mononoke/features/async_requests/types:async_requests_types",
         "//eden/mononoke/features/async_requests/worker_lib:worker_lib",
         "//eden/mononoke/megarepo_api:megarepo_api",
         "//eden/mononoke/metaconfig:metaconfig_types",

eden/mononoke/features/async_requests/worker/Cargo.toml

@@ -12,6 +12,7 @@ anyhow = "1.0.102"
 async-trait = "0.1.86"
 async_requests = { version = "0.1.0", path = ".." }
 async_requests_client = { version = "0.1.0", path = "../client" }
+async_requests_types = { version = "0.1.0", path = "../types" }
 blobstore = { version = "0.1.0", path = "../../../blobstore" }
 clap = { version = "4.6.0", features = ["derive", "env", "string", "unicode", "wrap_help"] }
 cmdlib_logging = { version = "0.1.0", path = "../../../cmdlib/log" }

eden/mononoke/features/async_requests/worker/src/main.rs

@@ -13,8 +13,10 @@ use anyhow::Context;
 use anyhow::Result;
 use async_requests::AsyncMethodRequestQueue;
 use async_requests::QueueRepoFilter;
+use async_requests::QueueRequestTypeFilter;
 use async_requests_client::open_blobstore;
 use async_requests_client::open_sql_connection;
+use async_requests_types::BACKFILL_REQUEST_TYPES;
 use async_trait::async_trait;
 use blobstore::Blobstore;
 use clap::Parser;
@@ -49,6 +51,17 @@ use worker_lib::worker::AsyncMethodRequestWorker;
 
 const SERVICE_NAME: &str = "async_requests_worker";
 
+/// Build a QueueRequestTypeFilter that excludes backfill request types.
+/// Backfill requests are handled by the dedicated backfill_worker.
+fn backfill_exclude_filter() -> QueueRequestTypeFilter {
+    QueueRequestTypeFilter::Except(
+        BACKFILL_REQUEST_TYPES
+            .iter()
+            .map(|s| requests_table::RequestType(s.to_string()))
+            .collect(),
+    )
+}
+
 const SM_CLEANUP_TIMEOUT_SECS: u64 = 60;
 
 /// Processes the megarepo async requests
@@ -122,10 +135,11 @@ impl RepoShardedProcess for WorkerProcess {
         let repos = vec![repo.repo_identity.id()];
         info!("Completed setup for repo {} ({:?})", repo_name, repos);
 
-        let queue = Arc::new(AsyncMethodRequestQueue::new(
+        let queue = Arc::new(AsyncMethodRequestQueue::new_with_request_type_filter(
             self.sql_connection.clone(),
             self.blobstore.clone(),
             QueueRepoFilter::Only(repos),
+            backfill_exclude_filter(),
         ));
 
         let executor = AsyncMethodRequestWorker::new(
@@ -296,10 +310,11 @@ fn run_worker_queue(
     will_exit: Arc<AtomicBool>,
 ) -> Result<()> {
     let executor = {
-        let queue = Arc::new(AsyncMethodRequestQueue::new(
+        let queue = Arc::new(AsyncMethodRequestQueue::new_with_request_type_filter(
             sql_connection,
             blobstore,
             repo_filter,
+            backfill_exclude_filter(),
         ));
 
         runtime.block_on(AsyncMethodRequestWorker::new(