aosp: implement streaming changelog on slapi

Summary:
In our last hackathon we worked on deprecating wireproto, but one of the big blockers was AOSP clones regression when we used commit graph segments. Therefore, we decided to migrate the streaming changelog endpoint over to SLAPI.

This is a first draft that mimics  what the wireproto endpoint was doing, still need to do the client side but figured I'd put this up for review for now since I'm working on this on my bits of free time.

Reviewed By: muirdm

Differential Revision: D77028873

fbshipit-source-id: afab86424ecd192071d42bfd62a1a8a405f664ac

Author: lmvasquezg

eden/mononoke/edenapi_service/BUCK

@@ -20,6 +20,7 @@ rust_library(
         "fbsource//third-party/rust:anyhow",
         "fbsource//third-party/rust:async-stream",
         "fbsource//third-party/rust:async-trait",
+        "//common/rust/shed/futures_ext:futures_ext",
         "fbsource//third-party/rust:base64",
         "fbsource//third-party/rust:bytes",
         "fbsource//third-party/rust:futures",
@@ -58,6 +59,7 @@ rust_library(
         "//common/rust/shed/stats:stats",
         "//common/rust/shed/time_ext:time_ext",
         "//eden/mononoke/repo_attributes/repo_blobstore:repo_blobstore",
+        "//eden/mononoke/repo_client:streaming_clone",
         "//eden/mononoke/blobstore:blobstore",
         "//eden/mononoke/blobstore:ephemeral_blobstore",
         "//eden/mononoke/bookmarks:bookmarks",

eden/mononoke/edenapi_service/Cargo.toml

@@ -26,6 +26,7 @@ ephemeral_blobstore = { version = "0.1.0", path = "../blobstore/ephemeral_blobst
 fbinit = { version = "0.2.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
 filestore = { version = "0.1.0", path = "../filestore" }
 futures = { version = "0.3.31", features = ["async-await", "compat"] }
+futures_ext = { version = "0.1.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
 futures_stats = { version = "0.1.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
 git_types = { version = "0.1.0", path = "../git/git_types" }
 gotham = "0.7.1"
@@ -71,6 +72,7 @@ slog = { package = "tracing_slog_compat", version = "0.1.0", git = "https://gith
 smallvec = { version = "1.15", features = ["impl_bincode", "serde", "specialization", "union"] }
 sorted_vector_map = { version = "0.2.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
 stats = { version = "0.1.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
+streaming_clone = { version = "0.1.0", path = "../repo_client/streaming_clone" }
 thiserror = "2.0.12"
 time_ext = { version = "0.1.0", git = "https://github.com/facebookexperimental/rust-shed.git", branch = "main" }
 time_window_counter = { version = "0.1.0", path = "../time_window_counter" }

eden/mononoke/edenapi_service/src/handlers/legacy.rs

@@ -5,16 +5,35 @@
  * GNU General Public License version 2.
  */
 
+use std::time::Duration;
+
 use async_trait::async_trait;
+use cloned::cloned;
+use context::PerfCounterType;
+use edenapi_types::ServerError;
 use edenapi_types::StreamingChangelogRequest;
 use edenapi_types::StreamingChangelogResponse;
+use edenapi_types::legacy::Metadata;
+use edenapi_types::legacy::StreamingChangelogBlob;
+use edenapi_types::legacy::StreamingChangelogData;
+use futures::future::FutureExt;
+use futures::stream;
+use futures::stream::StreamExt;
+use futures::stream::TryStreamExt;
+use futures_ext::FbStreamExt;
+use futures_stats::TimedFutureExt;
 use mononoke_api::Repo;
+use slog::debug;
+use streaming_clone::StreamingCloneArc;
+use time_ext::DurationExt;
 
 use super::HandlerResult;
 use super::SaplingRemoteApiHandler;
 use super::SaplingRemoteApiMethod;
 use super::handler::SaplingRemoteApiContext;
 
+const TIMEOUT_SECS: Duration = Duration::from_secs(4 * 60 * 60);
+
 /// Legacy streaming changelog handler from wireproto.
 #[allow(dead_code)]
 pub struct StreamingCloneHandler;
@@ -29,9 +48,108 @@ impl SaplingRemoteApiHandler for StreamingCloneHandler {
     const ENDPOINT: &'static str = "/streaming_clone";
 
     async fn handler(
-        _ectx: SaplingRemoteApiContext<Self::PathExtractor, Self::QueryStringExtractor, Repo>,
-        _request: Self::Request,
+        ectx: SaplingRemoteApiContext<Self::PathExtractor, Self::QueryStringExtractor, Repo>,
+        request: Self::Request,
     ) -> HandlerResult<'async_trait, Self::Response> {
-        unimplemented!("StreamingCloneHandler is not implemented")
+        let streaming_clone = ectx.repo().repo().streaming_clone_arc();
+        let ctx = ectx.repo().ctx().clone();
+
+        let changelog = streaming_clone
+            .fetch_changelog(ctx.clone(), request.tag.as_deref())
+            .await?;
+
+        let data_blob_chunk_count = 0;
+        let data_blobs: Vec<_> = changelog
+            .data_blobs
+            .into_iter()
+            .map(|fut| {
+                cloned!(ctx);
+                async move {
+                    let (stats, res) = fut.timed().await;
+                    ctx.perf_counters().add_to_counter(
+                        PerfCounterType::SumManifoldPollTime,
+                        stats.poll_time.as_nanos_unchecked() as i64,
+                    );
+                    if let Ok(bytes) = res.as_ref() {
+                        ctx.perf_counters()
+                            .add_to_counter(PerfCounterType::BytesSent, bytes.len() as i64)
+                    }
+
+                    let data = res.map(|res| {
+                        StreamingChangelogData::DataBlobChunk(StreamingChangelogBlob {
+                            chunk: res.into(),
+                            chunk_id: data_blob_chunk_count,
+                        })
+                    });
+
+                    StreamingChangelogResponse {
+                        data: data.map_err(|e| ServerError::generic(format!("{:?}", e))),
+                    }
+                }
+                .boxed()
+            })
+            .collect();
+
+        let index_blob_chunk_count = 0;
+        let index_blobs: Vec<_> = changelog
+            .index_blobs
+            .into_iter()
+            .map(|fut| {
+                cloned!(ctx);
+                async move {
+                    let (stats, res) = fut.timed().await;
+                    ctx.perf_counters().add_to_counter(
+                        PerfCounterType::SumManifoldPollTime,
+                        stats.poll_time.as_nanos_unchecked() as i64,
+                    );
+                    if let Ok(bytes) = res.as_ref() {
+                        ctx.perf_counters()
+                            .add_to_counter(PerfCounterType::BytesSent, bytes.len() as i64)
+                    }
+
+                    let data = res.map(|res| {
+                        StreamingChangelogData::IndexBlobChunk(StreamingChangelogBlob {
+                            chunk: res.into(),
+                            chunk_id: index_blob_chunk_count,
+                        })
+                    });
+
+                    StreamingChangelogResponse {
+                        data: data.map_err(|e| ServerError::generic(format!("{:?}", e))),
+                    }
+                }
+                .boxed()
+            })
+            .collect();
+
+        debug!(
+            ctx.logger(),
+            "streaming changelog {} index bytes, {} data bytes",
+            changelog.index_size,
+            changelog.data_size
+        );
+
+        let metadata = StreamingChangelogData::Metadata(Metadata {
+            index_size: changelog.index_size as u64,
+            data_size: changelog.data_size as u64,
+        });
+        let mut response_header = Vec::new();
+        response_header.push(metadata);
+
+        let response = stream::iter(
+            response_header
+                .into_iter()
+                .map(|data| StreamingChangelogResponse { data: Ok(data) }),
+        );
+
+        let res = response
+            .chain(stream::iter(index_blobs).buffered(100))
+            .chain(stream::iter(data_blobs).buffered(100));
+
+        Ok(res
+            .whole_stream_timeout(TIMEOUT_SECS)
+            .yield_periodically()
+            .map_err(|e| e.into())
+            .boxed())
     }
 }

eden/scm/lib/edenapi/types/src/legacy.rs

@@ -50,19 +50,19 @@ impl Default for StreamingChangelogData {
 #[derive(Clone, Debug, Deserialize, Serialize, Eq, PartialEq)]
 pub struct StreamingChangelogBlob {
     #[id(0)]
-    chunk: Bytes,
+    pub chunk: Bytes,
     #[id(1)]
-    chunk_id: u64,
+    pub chunk_id: u64,
 }
 
 #[auto_wire]
 #[derive(Clone, Debug, Eq, PartialEq, Serialize, Deserialize)]
 #[cfg_attr(any(test, feature = "for-tests"), derive(Arbitrary))]
 pub struct Metadata {
     #[id(0)]
-    index_size: u64,
+    pub index_size: u64,
     #[id(1)]
-    data_size: u64,
+    pub data_size: u64,
 }
 
 #[cfg(any(test, feature = "for-tests"))]