admin: add inspect subcommand to derivation-queue

Summary:
Add an `inspect` subcommand to `monad derivation-queue` that shows the Zelos DAG state for a specific queue item. This helps debug stuck or broken items by showing:
- Whether the `needed` node exists and its metadata (retry count, priority, enqueue time, head changeset)
- Whether the item is in the `ready` or `deriving` state
- Forward dependencies and whether each dependency's `needed` node still exists (marks broken deps)
- Reverse dependencies (items blocked on this one)

Reviewed By: lmvasquezg

Differential Revision: D103875695

fbshipit-source-id: d25eaffe042b896a0e838aacfc7404f9cbce2dff

Author: YousefSalama

eden/mononoke/repo_attributes/repo_derivation_queues/src/lib.rs

@@ -49,6 +49,28 @@ pub use crate::dag_items::derivation_priority_to_str;
 pub use crate::errors::InternalError;
 pub use crate::underived::build_underived_batched_graph;
 
+/// Status of a single dependency: suffix and whether its `needed` node exists.
+pub struct DepStatus {
+    pub suffix: String,
+    pub needed_exists: bool,
+}
+
+/// Whether a queue item is in the ready state, and which priority queue.
+pub enum ReadyState {
+    NotReady,
+    ReadyHighPri,
+    ReadyLowPri,
+}
+
+/// Result of inspecting a specific DAG item in the derivation queue.
+pub struct InspectResult {
+    pub needed: Option<DagItemInfo>,
+    pub ready_state: ReadyState,
+    pub is_deriving: bool,
+    pub forward_deps: Vec<DepStatus>,
+    pub reverse_deps: Vec<DepStatus>,
+}
+
 /// Lightweight item returned from dequeue — just the ID and priority.
 /// The full DerivationDagItem is constructed during claim_derivation
 /// after fetching data from Zeus.
@@ -155,6 +177,14 @@ pub trait DerivationQueue {
 
     async fn summary(&self, ctx: &CoreContext) -> Result<DerivationQueueSummary, InternalError>;
 
+    /// Inspect the Zelos DAG state of a specific item, including its node
+    /// states, forward dependencies, and reverse dependencies.
+    async fn inspect(
+        &self,
+        ctx: &CoreContext,
+        item_id: DagItemId,
+    ) -> Result<InspectResult, InternalError>;
+
     fn derived_data_manager(&self) -> &DerivedDataManager;
 
     fn repo_id(&self) -> RepositoryId;

eden/mononoke/tools/admin/src/commands/derivation_queue.rs

@@ -6,6 +6,7 @@
  */
 
 mod enqueue;
+mod inspect;
 mod summary;
 mod unsafe_evict;
 
@@ -18,6 +19,7 @@ use bookmarks::Bookmarks;
 use clap::Parser;
 use clap::Subcommand;
 use enqueue::EnqueueArgs;
+use inspect::InspectArgs;
 use metaconfig_types::RepoConfig;
 use metaconfig_types::RepoConfigRef;
 use mononoke_app::MononokeApp;
@@ -49,6 +51,8 @@ pub enum DerivationQueueSubcommand {
     Summary(SummaryArgs),
     /// Evict an item (referenced by root cs_id and derived data type) from the derivation queue. WARNING: can leave dependent items in the queue stuck
     UnsafeEvict(UnsafeEvictArgs),
+    /// Inspect the Zelos DAG state of a specific item in the derivation queue
+    Inspect(InspectArgs),
 }
 
 #[facet::container]
@@ -97,5 +101,8 @@ pub async fn run(app: MononokeApp, args: CommandArgs) -> Result<()> {
         DerivationQueueSubcommand::UnsafeEvict(args) => {
             unsafe_evict::unsafe_evict(&ctx, &repo, config_name, args).await
         }
+        DerivationQueueSubcommand::Inspect(args) => {
+            inspect::inspect(&ctx, &repo, config_name, args).await
+        }
     }
 }

eden/mononoke/tools/admin/src/commands/derivation_queue/inspect.rs

@@ -0,0 +1,129 @@
+/*
+ * Copyright (c) Meta Platforms, Inc. and affiliates.
+ *
+ * This software may be used and distributed according to the terms of the
+ * GNU General Public License version 2.
+ */
+
+use anyhow::Result;
+use anyhow::anyhow;
+use clap::Args;
+use context::CoreContext;
+use mononoke_app::args::ChangesetArgs;
+use mononoke_app::args::DerivedDataArgs;
+use repo_derivation_queues::DagItemId;
+use repo_derivation_queues::ReadyState;
+use repo_derivation_queues::RepoDerivationQueuesRef;
+use repo_derivation_queues::derivation_priority_to_str;
+use repo_identity::RepoIdentityRef;
+
+use super::Repo;
+
+#[derive(Args)]
+pub struct InspectArgs {
+    #[clap(flatten)]
+    changeset_args: ChangesetArgs,
+
+    #[clap(flatten)]
+    derived_data_args: DerivedDataArgs,
+
+    /// Stage ID for staged derivation items
+    #[clap(long)]
+    stage_id: Option<String>,
+}
+
+pub async fn inspect(
+    ctx: &CoreContext,
+    repo: &Repo,
+    config_name: &str,
+    args: InspectArgs,
+) -> Result<()> {
+    let derivation_queue = repo
+        .repo_derivation_queues()
+        .queue(config_name)
+        .ok_or_else(|| anyhow!("Missing derivation queue for config {}", config_name))?;
+
+    let derived_data_type = args.derived_data_args.resolve_type()?;
+    let cs_ids = args.changeset_args.resolve_changesets(ctx, repo).await?;
+
+    for cs_id in cs_ids {
+        let item_id = DagItemId::new(
+            repo.repo_identity().id(),
+            config_name.to_string(),
+            derived_data_type,
+            cs_id,
+            args.stage_id.clone(),
+        );
+
+        println!(
+            "Item: {:?}/{}/{}",
+            derived_data_type,
+            args.stage_id.as_deref().unwrap_or("(no stage)"),
+            cs_id
+        );
+
+        let result = derivation_queue.inspect(ctx, item_id).await?;
+
+        match &result.needed {
+            Some(info) => {
+                let ts = info
+                    .enqueue_timestamp()
+                    .map(|t| format!("{}s{}ms ago", t.since_seconds(), t.since_millis() % 1000))
+                    .unwrap_or_else(|| "unknown".to_string());
+                println!(
+                    "  needed:   EXISTS (retry_count={}, priority={}, enqueued {})",
+                    info.retry_count(),
+                    derivation_priority_to_str(info.priority()),
+                    ts
+                );
+                println!("  head:     {}", info.head_cs_id());
+            }
+            None => println!("  needed:   MISSING"),
+        }
+
+        let ready_str = match result.ready_state {
+            ReadyState::NotReady => "no",
+            ReadyState::ReadyHighPri => "YES (high priority)",
+            ReadyState::ReadyLowPri => "YES (low priority)",
+        };
+        println!("  ready:    {}", ready_str);
+        println!(
+            "  deriving: {}",
+            if result.is_deriving { "YES" } else { "no" }
+        );
+
+        if result.forward_deps.is_empty() {
+            println!("  forward deps: (none)");
+        } else {
+            println!("  forward deps ({}):", result.forward_deps.len());
+            for dep in &result.forward_deps {
+                let status = if dep.needed_exists {
+                    "EXISTS"
+                } else {
+                    "MISSING"
+                };
+                let marker = if !dep.needed_exists { " <- BROKEN" } else { "" };
+                println!("    {} -> needed: {}{}", dep.suffix, status, marker);
+            }
+        }
+
+        if result.reverse_deps.is_empty() {
+            println!("  reverse deps: (none)");
+        } else {
+            println!("  reverse deps ({}):", result.reverse_deps.len());
+            for dep in &result.reverse_deps {
+                let status = if dep.needed_exists {
+                    "EXISTS"
+                } else {
+                    "MISSING"
+                };
+                let marker = if !dep.needed_exists { " <- BROKEN" } else { "" };
+                println!("    {} -> needed: {}{}", dep.suffix, status, marker);
+            }
+        }
+
+        println!();
+    }
+
+    Ok(())
+}