lfs_server: gate upstream batch_download fetch behind a JustKnob

Summary:
Adds `scm/mononoke:lfs_server_skip_upstream_for_downloads`, a per-repo JustKnob that, when enabled, makes `batch_download` skip the upstream LFS server entirely. Only the internal Filestore is consulted; objects that exist solely upstream are reported as 404 to the client. Upload paths (`batch_upload`, `upload_from_client`, `sync_internal_and_upstream`) are unaffected and still propagate uploads to upstream.

The motivation is to support repos where new content has stopped being mirrored upstream: clients should still write through to upstream so the mirror stays in sync, but downloads should not depend on or wait for upstream.

The existing knob `repo_config().lfs.use_upstream_lfs_server` is all-or-nothing — disabling it removes upstream from both directions. The new JK is download-only and switchable per-repo (via `switchval = repo_name`) so the change can be canaried instead of flipped globally.

Default in `just_knobs.json` is `false`, preserving existing behavior.

Reviewed By: YousefSalama

Differential Revision: D104026975

fbshipit-source-id: d82f01df50da3a5adb4e618668f029a16fb49a5f

Author: mzr

eden/mononoke/common/mononoke_macros/just_knobs_defaults/just_knobs.json

@@ -69,6 +69,7 @@
     "scm/mononoke:use_acl_manifest_for_restricted_paths": true,
     "scm/metagit:mononoke_git_lfs": false,
     "scm/mononoke:lfs_server_compression_sniff_enabled": false,
+    "scm/mononoke:lfs_server_skip_upstream_for_downloads": false,
     "scm/mononoke:use_split_config_loading": false,
     "scm/mononoke:cross_repo_pause_backsyncer": false,
     "scm/mononoke:allow_bare_author_unixname": true,

eden/mononoke/servers/lfs/lfs_server/src/batch.rs

@@ -49,6 +49,7 @@ use mononoke_types::hash::Sha256;
 use mononoke_types::typed_hash::ContentId;
 use redactedblobstore::has_redaction_root_cause;
 use repo_blobstore::RepoBlobstoreRef;
+use repo_identity::RepoIdentityRef;
 use serde::Deserialize;
 use stats::prelude::*;
 use time_ext::DurationExt;
@@ -73,6 +74,13 @@ define_stats! {
     upload_rejected: timeseries(Rate, Sum),
 }
 
+/// JustKnob that, when enabled for a repo, makes batch download responses
+/// ignore the upstream LFS server entirely. Internal-only objects are returned
+/// as usual; objects that exist only upstream become 404s. Upload paths are
+/// unaffected. Per `eden/.llms/rules/rust_unwrap_safety.md`, a bare `?` is the
+/// correct pattern here; defaults belong in `just_knobs.json`.
+const SKIP_UPSTREAM_FOR_DOWNLOADS_JK: &str = "scm/mononoke:lfs_server_skip_upstream_for_downloads";
+
 enum Source {
     Internal,
     Upstream,
@@ -566,6 +574,25 @@ async fn batch_download(
     batch: RequestBatch,
     scuba: &mut Option<&mut ScubaMiddlewareState>,
 ) -> Result<ResponseBatch, ErrorKind> {
+    if justknobs::eval(
+        SKIP_UPSTREAM_FOR_DOWNLOADS_JK,
+        None,
+        Some(ctx.repo.repo_identity().name()),
+    )
+    .map_err(ErrorKind::Error)?
+    {
+        let internal_objects = internal_objects(ctx, &batch.objects).await?;
+        ScubaMiddlewareState::maybe_add(scuba, LfsScubaKey::BatchOrder, "skip_upstream");
+        let upstream = Ok(UpstreamObjects::NoUpstream);
+        let objects =
+            batch_download_response_objects(&batch.objects, &upstream, &internal_objects, scuba)
+                .map_err(ErrorKind::Error)?;
+        return Ok(ResponseBatch {
+            transfer: Transfer::Basic,
+            objects,
+        });
+    }
+
     let upstream = upstream_objects(ctx, &batch.objects).fuse();
     let internal = internal_objects(ctx, &batch.objects).fuse();
     pin_mut!(upstream, internal);
@@ -1207,4 +1234,49 @@ mod test {
 
         Ok(())
     }
+
+    #[mononoke::fbinit_test]
+    async fn test_batch_download_skips_upstream_when_jk_enabled(
+        fb: FacebookInit,
+    ) -> Result<(), Error> {
+        use justknobs::test_helpers::JustKnobsInMemory;
+        use justknobs::test_helpers::KnobVal;
+        use justknobs::test_helpers::with_just_knobs_async;
+
+        // Build a repo whose LFS config enables upstream and point the context
+        // at an upstream URI. The test HTTP client is Disabled — so any actual
+        // dispatch to upstream would panic. The test passes only if the JK gate
+        // prevents that call.
+        let repo: Repo = TestRepoFactory::new(fb)?
+            .with_config_override(|c| c.lfs.use_upstream_lfs_server = true)
+            .build()
+            .await?;
+
+        let ctx = RepositoryRequestContext::test_builder_with_repo(fb, repo)?
+            .upstream_uri(Some("http://upstream.example/".to_string()))
+            .build()?;
+
+        let request = RequestBatch {
+            operation: Operation::Download,
+            r#ref: None,
+            transfers: vec![Transfer::Basic],
+            objects: vec![obj(ONES_SHA256, 111)],
+        };
+
+        let res = with_just_knobs_async(
+            JustKnobsInMemory::new(hashmap! {
+                SKIP_UPSTREAM_FOR_DOWNLOADS_JK.to_string() => KnobVal::Bool(true),
+            }),
+            Box::pin(batch_download(&ctx, request, &mut None)),
+        )
+        .await?;
+
+        assert_eq!(res.objects.len(), 1);
+        match &res.objects[0].status {
+            ObjectStatus::Err { error } => assert_eq!(error.code, 404),
+            ObjectStatus::Ok { .. } => panic!("expected 404, got Ok"),
+        }
+
+        Ok(())
+    }
 }