Delete dewey_res.bzl

Summary:
Delete tools/build_defs/android/dewey_res.bzl which provides the dewey_custom_res macro.

Dewey backing data was removed on April 1. The only two remaining consumers (oxygen zip and verification tests) are being cleaned up in D99099121 and D99099398. This diff removes the macro definition itself.

Part of the Dewey deprecation cleanup.

Reviewed By: YousefSalama

Differential Revision: D99820098

fbshipit-source-id: bf58ea7fb287d84ba725e1a9903165c4feed2898

Author: Clara Rull

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

@@ -122,7 +122,8 @@ impl RepoShardedProcess for WorkerProcess {
             .add_repo(repo_name)
             .await
             .with_context(|| format!("Failure in setting up repo {}", repo_name))?;
-        let repos = vec![repo.repo_identity.id()];
+        let repo_id = repo.repo_identity.id();
+        let repos = vec![repo_id];
         info!("Completed setup for repos {:?}", repos);
 
         let queue = Arc::new(AsyncMethodRequestQueue::new(
@@ -138,6 +139,7 @@ impl RepoShardedProcess for WorkerProcess {
             self.mononoke.clone(),
             self.megarepo.clone(),
             self.will_exit.clone(),
+            Some(repo_id),
         )
         .await?;
         Ok(Arc::new(executor))
@@ -311,6 +313,7 @@ fn run_worker_queue(
             mononoke.clone(),
             megarepo.clone(),
             will_exit.clone(),
+            None, // non-sharded: no specific repo_id
         ))?
     };
     runtime.spawn(async move { executor.execute().await });

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

@@ -5,15 +5,13 @@
  * GNU General Public License version 2.
  */
 
-use std::collections::HashMap;
 use std::time::Duration;
 
 use async_requests::AsyncMethodRequestQueue;
 use async_requests::types::RequestStatus;
 use context::CoreContext;
 use mononoke_api::RepositoryId;
 use mononoke_types::Timestamp;
-use requests_table::QueueStatsEntry;
 use stats::define_stats;
 use stats::prelude::*;
 use tracing::warn;
@@ -37,9 +35,19 @@ define_stats! {
     queue_age_by_repo_and_status: dynamic_singleton_counter("queue.{}.age_s.{}", (repo_id: String, status: String)),
 }
 
+/// Report queue stats for this shard's repo only.
+///
+/// Each shard reports metrics only for its own repo_id. This avoids
+/// cross-shard racing on shared counters and prevents stale values
+/// when a shard is reassigned: the counter simply goes absent (no
+/// data) rather than being stuck at a stale non-zero value. A dead
+/// shard detector should alert on absence of data.
+///
+/// For the non-sharded code path, `repo_id` is `None` and only the
+/// global (non-per-repo) counters are reported.
 pub(crate) async fn stats_loop(
     ctx: &CoreContext,
-    repo_ids: Vec<RepositoryId>,
+    repo_id: Option<RepositoryId>,
     queue: &AsyncMethodRequestQueue,
 ) {
     loop {
@@ -49,8 +57,10 @@ pub(crate) async fn stats_loop(
             Ok(res) => {
                 process_queue_length_by_status(ctx, &res);
                 process_queue_age_by_status(ctx, now, &res);
-                process_queue_length_by_repo_and_status(ctx, &repo_ids, &res);
-                process_queue_age_by_repo_and_status(ctx, &repo_ids, now, &res);
+                if let Some(repo_id) = &repo_id {
+                    process_queue_length_by_repo(ctx, repo_id, &res);
+                    process_queue_age_by_repo(ctx, repo_id, now, &res);
+                }
             }
             Err(err) => {
                 STATS::stats_error.add_value(1);
@@ -62,51 +72,22 @@ pub(crate) async fn stats_loop(
     }
 }
 
-// Keep track of the stats we have already logged. Any missing ones, we will log a 0.
-struct Seen {
-    inner: HashMap<QueueStatsEntry, bool>,
-}
-
-impl Seen {
-    fn new(repo_ids: &Vec<RepositoryId>) -> Self {
-        let mut seen = HashMap::new();
-        for status in STATUSES {
-            for repo_id in repo_ids {
-                seen.insert(
-                    QueueStatsEntry {
-                        repo_id: Some(repo_id.clone()),
-                        status,
-                    },
-                    false,
-                );
-            }
-        }
-        Self { inner: seen }
-    }
-
-    fn mark(&mut self, repo_id: Option<RepositoryId>, status: RequestStatus) {
-        self.inner.insert(QueueStatsEntry { repo_id, status }, true);
-    }
-
-    fn get_missing(&self) -> Vec<QueueStatsEntry> {
-        self.inner
-            .iter()
-            .filter_map(|(entry, seen)| if !*seen { Some(entry) } else { None })
-            .cloned()
-            .collect()
-    }
-}
-
 fn process_queue_length_by_status(ctx: &CoreContext, res: &requests_table::QueueStats) {
-    let mut seen = Seen::new(&vec![]);
+    // Report whatever the DB returns for global stats, then zero out
+    // any of the 4 statuses that were absent from the result.
+    let mut seen = [false; 4];
     let stats = &res.queue_length_by_status;
     for (status, count) in stats {
-        seen.mark(None, *status);
+        if let Some(idx) = status_index(status) {
+            seen[idx] = true;
+        }
         STATS::queue_length_by_status.set_value(ctx.fb, *count as i64, (status.to_string(),));
     }
 
-    for entry in seen.get_missing() {
-        STATS::queue_length_by_status.set_value(ctx.fb, 0, (entry.status.to_string(),));
+    for (idx, was_seen) in seen.iter().enumerate() {
+        if !was_seen {
+            STATS::queue_length_by_status.set_value(ctx.fb, 0, (STATUSES[idx].to_string(),));
+        }
     }
 }
 
@@ -115,79 +96,93 @@ fn process_queue_age_by_status(
     now: Timestamp,
     res: &requests_table::QueueStats,
 ) {
-    let mut seen = Seen::new(&vec![]);
+    let mut seen = [false; 4];
     let stats = &res.queue_age_by_status;
     for (status, ts) in stats {
-        seen.mark(None, *status);
+        if let Some(idx) = status_index(status) {
+            seen[idx] = true;
+        }
         let diff = std::cmp::max(now.timestamp_seconds() - ts.timestamp_seconds(), 0);
         STATS::queue_age_by_status.set_value(ctx.fb, diff, (status.to_string(),));
     }
 
-    for entry in seen.get_missing() {
-        STATS::queue_age_by_status.set_value(ctx.fb, 0, (entry.status.to_string(),));
+    for (idx, was_seen) in seen.iter().enumerate() {
+        if !was_seen {
+            STATS::queue_age_by_status.set_value(ctx.fb, 0, (STATUSES[idx].to_string(),));
+        }
     }
 }
 
-fn process_queue_length_by_repo_and_status(
+/// Report per-repo queue length for this shard's repo only.
+/// If the DB returns no data for a status, we report 0.
+fn process_queue_length_by_repo(
     ctx: &CoreContext,
-    repo_ids: &Vec<RepositoryId>,
+    repo_id: &RepositoryId,
     res: &requests_table::QueueStats,
 ) {
-    let mut seen = Seen::new(repo_ids);
+    let mut seen = [false; 4];
+    let repo_id_str = repo_id.to_string();
     let stats = &res.queue_length_by_repo_and_status;
     for (entry, count) in stats {
-        seen.mark(entry.repo_id, entry.status);
-        STATS::queue_length_by_repo_and_status.set_value(
-            ctx.fb,
-            *count as i64,
-            (
-                entry.repo_id.unwrap_or(RepositoryId::new(0)).to_string(),
-                entry.status.to_string(),
-            ),
-        );
+        if entry.repo_id.as_ref() == Some(repo_id) {
+            if let Some(idx) = status_index(&entry.status) {
+                seen[idx] = true;
+            }
+            STATS::queue_length_by_repo_and_status.set_value(
+                ctx.fb,
+                *count as i64,
+                (repo_id_str.clone(), entry.status.to_string()),
+            );
+        }
     }
 
-    for entry in seen.get_missing() {
-        STATS::queue_length_by_repo_and_status.set_value(
-            ctx.fb,
-            0,
-            (
-                entry.repo_id.unwrap_or(RepositoryId::new(0)).to_string(),
-                entry.status.to_string(),
-            ),
-        );
+    for (idx, was_seen) in seen.iter().enumerate() {
+        if !was_seen {
+            STATS::queue_length_by_repo_and_status.set_value(
+                ctx.fb,
+                0,
+                (repo_id_str.clone(), STATUSES[idx].to_string()),
+            );
+        }
     }
 }
 
-fn process_queue_age_by_repo_and_status(
+/// Report per-repo queue age for this shard's repo only.
+/// If the DB returns no data for a status, we report 0.
+fn process_queue_age_by_repo(
     ctx: &CoreContext,
-    repo_ids: &Vec<RepositoryId>,
+    repo_id: &RepositoryId,
     now: Timestamp,
     res: &requests_table::QueueStats,
 ) {
-    let mut seen = Seen::new(repo_ids);
+    let mut seen = [false; 4];
+    let repo_id_str = repo_id.to_string();
     let stats = &res.queue_age_by_repo_and_status;
     for (entry, ts) in stats {
-        seen.mark(entry.repo_id, entry.status);
-        let diff = std::cmp::max(now.timestamp_seconds() - ts.timestamp_seconds(), 0);
-        STATS::queue_age_by_repo_and_status.set_value(
-            ctx.fb,
-            diff,
-            (
-                entry.repo_id.unwrap_or(RepositoryId::new(0)).to_string(),
-                entry.status.to_string(),
-            ),
-        );
+        if entry.repo_id.as_ref() == Some(repo_id) {
+            if let Some(idx) = status_index(&entry.status) {
+                seen[idx] = true;
+            }
+            let diff = std::cmp::max(now.timestamp_seconds() - ts.timestamp_seconds(), 0);
+            STATS::queue_age_by_repo_and_status.set_value(
+                ctx.fb,
+                diff,
+                (repo_id_str.clone(), entry.status.to_string()),
+            );
+        }
     }
 
-    for entry in seen.get_missing() {
-        STATS::queue_age_by_repo_and_status.set_value(
-            ctx.fb,
-            0,
-            (
-                entry.repo_id.unwrap_or(RepositoryId::new(0)).to_string(),
-                entry.status.to_string(),
-            ),
-        );
+    for (idx, was_seen) in seen.iter().enumerate() {
+        if !was_seen {
+            STATS::queue_age_by_repo_and_status.set_value(
+                ctx.fb,
+                0,
+                (repo_id_str.clone(), STATUSES[idx].to_string()),
+            );
+        }
     }
 }
+
+fn status_index(status: &RequestStatus) -> Option<usize> {
+    STATUSES.iter().position(|s| s == status)
+}

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

@@ -42,8 +42,8 @@ use futures_stats::TimedFutureExt;
 use hostname::get_hostname;
 use megarepo_api::MegarepoApi;
 use mononoke_api::Mononoke;
-use mononoke_api::MononokeRepo;
 use mononoke_api::Repo;
+use mononoke_api::RepositoryId;
 use mononoke_macros::mononoke;
 use mononoke_types::Timestamp;
 use stats::define_stats;
@@ -86,6 +86,8 @@ pub struct AsyncMethodRequestWorker {
     will_exit: Arc<AtomicBool>,
     limit: Option<usize>,
     concurrency_limit: usize,
+    /// The repo this shard is responsible for. `None` for non-sharded workers.
+    repo_id: Option<RepositoryId>,
 }
 
 impl AsyncMethodRequestWorker {
@@ -96,6 +98,7 @@ impl AsyncMethodRequestWorker {
         mononoke: Arc<Mononoke<Repo>>,
         megarepo: Arc<MegarepoApi<Repo>>,
         will_exit: Arc<AtomicBool>,
+        repo_id: Option<RepositoryId>,
     ) -> Result<Self, Error> {
         let name = {
             let tw_job_cluster = std::env::var("TW_JOB_CLUSTER");
@@ -121,6 +124,7 @@ impl AsyncMethodRequestWorker {
             will_exit,
             limit: args.request_limit,
             concurrency_limit: args.jobs,
+            repo_id,
         })
     }
 }
@@ -133,11 +137,13 @@ impl RepoShardedProcessExecutor for AsyncMethodRequestWorker {
     /// will_exit atomic bool is a flag to prevent the worker from grabbing new
     /// items from the queue and gracefully terminate.
     async fn execute(&self) -> Result<()> {
-        // Start the stats logger loop
+        // Start the stats logger loop.
+        // Each shard reports metrics only for its own repo to avoid
+        // cross-shard counter races and stale values on shard reassignment.
         let (stats, stats_abort_handle) = abortable({
             cloned!(self.ctx, self.queue);
-            let repo_ids = self.mononoke.known_repo_ids().clone();
-            async move { stats_loop(&ctx, repo_ids, &queue).await }
+            let repo_id = self.repo_id;
+            async move { stats_loop(&ctx, repo_id, &queue).await }
         });
         let _stats = mononoke::spawn_task(stats);