pmeunier/elpe - Change AJLMC7UMTMBYBXLOSMGT5PDNX4D7O7LV4GDV7AN25CPOBPRUFYVAC

Forwarding log to frontend

Created by pmeunier on July 2, 2025

AJLMC7UMTMBYBXLOSMGT5PDNX4D7O7LV4GDV7AN25CPOBPRUFYVAC

Dependencies

In channels

main

Change contents

Insertion in src/main.rs at line 9 [5.1363]
[3.44]
[4.1587]
```
use tokio_util::sync::PollSender;
```
Insertion in src/main.rs at line 38 [5.1363]
[4.2038]
[4.2038]
```
        debug!("add_path");
```

Replacement in src/main.rs at line 41 [5.1363]

∅:D[6.387] → [4.2127:2179]

B:BD[4.2127] → [4.2127:2179]

        let tmp_dir = tempfile::tempdir_in(store)?;

[6.387]

[4.2179]

        debug!("store");
        let tmp_dir = tempfile::tempdir_in(store).unwrap();
        debug!("store {:?}", tmp_dir);

Replacement in src/main.rs at line 46 [5.1363]
∅:D[4.2228] → [5.2000:2001]
B:BD[5.2000] → [5.2000:2001]
[4.2228]
[4.2229]
```
        debug!("loop");
```

Insertion in src/main.rs at line 53 [5.1363]

[4.2482]

                    info!("Adding file {:?}", f.name);

Insertion in src/main.rs at line 62 [5.1363]

[4.2922]

                    info!("Adding file {:?}", d.name);

Insertion in src/main.rs at line 76 [5.1363]
[4.3477]
[4.3477]
```
        debug!("loop done");
```
Insertion in src/main.rs at line 220 [5.1363]
[5.2951]
[5.2951]
```
    type UbuntuPackageStream = ResponseStream;
```

Replacement in src/main.rs at line 225 [5.1363]

B:BD[4.6060] → [4.6060:6134]

    ) -> Result<tonic::Response<proto::DerivationReply>, tonic::Status> {

[4.6060]

[5.3117]

    ) -> Result<tonic::Response<Self::UbuntuPackageStream>, tonic::Status> {

Replacement in src/main.rs at line 245 [5.1363]

B:BD[2.294] → [2.294:380]

∅:D[2.380] → [5.3465:3507]

B:BD[5.3465] → [5.3465:3507]

B:BD[5.3507] → [4.6135:6180]

∅:D[4.6180] → [5.3538:3595]

B:BD[5.3538] → [5.3538:3595]

B:BD[5.3595] → [4.6181:6646]

        let p = download_extract_deps(&index, &self.deb_client, &r.name, &link_extra)
            .await
            .unwrap();
        info!("path {:?} {:#?}", r.name, p);
        Ok(tonic::Response::new(proto::DerivationReply {
            result: Some(proto::derivation_reply::Result::Ok(
                proto::DerivationResult {
                    destdir: p
                        .result
                        .iter()
                        .rev()
                        .map(|x| x.to_str().unwrap().to_string())
                        .collect(),
                    paths: p.paths.into_iter().filter_map(Arc::into_inner).collect(),
                    path_patterns: Vec::new(),

[2.294]

[4.6646]


        let (tx, rx) = tokio::sync::mpsc::channel(200);
        use crate::extract::Msg;
        let output_stream = ReceiverStream::new(rx).map(|x| {
            Ok(match x {
                Msg::Downloading(p) => proto::DerivationReply {
                    result: Some(proto::derivation_reply::Result::Loading(p)),

Replacement in src/main.rs at line 253 [5.1363]

B:BD[4.6665] → [4.6665:6681]

∅:D[4.6681] → [5.3773:3785]

B:BD[5.3773] → [5.3773:3785]

            )),
        }))

[4.6665]

[5.4845]

                Msg::Ok(p) => proto::DerivationReply {
                    result: Some(proto::derivation_reply::Result::Ok(
                        proto::DerivationResult {
                            destdir: p
                                .result
                                .iter()
                                .rev()
                                .map(|x| x.to_str().unwrap().to_string())
                                .collect(),
                            paths: p.paths.into_iter().filter_map(Arc::into_inner).collect(),
                            path_patterns: Vec::new(),
                        },
                    )),
                },
                Msg::Error(e) => proto::DerivationReply {
                    result: Some(proto::derivation_reply::Result::Error(e.to_string())),
                },
            })
        });
        match download_extract_deps(
            &index,
            &self.deb_client,
            &r.name,
            &link_extra,
            PollSender::new(tx.clone()),
        )
        .await
        {
            Ok(p) => {
                info!("path {:?} {:#?}", r.name, p);
                tx.send(Msg::Ok(p)).await.unwrap();
            }
            Err(e) => tx.send(Msg::Error(e)).await.unwrap(),
        }
        Ok(tonic::Response::new(Box::pin(output_stream)))

Insertion in src/lib.rs at line 10 [5.10326]
[4.8271]
[5.10327]
```
use futures::SinkExt;
```
Insertion in src/lib.rs at line 36 [5.10326]
[4.8732]
[5.10537]
```
mod dummy_sink;
use dummy_sink::*;
```

Insertion in src/lib.rs at line 93 [5.10326]

[4.10103]

[5.11210]

    /// Package not found
    #[error("Package not found: {pkg}")]
    PackageNotFound {
        /// The package name.
        pkg: String,
    },

Insertion in src/lib.rs at line 311 [5.10326]

[6.1497]

[4.11035]

        std::fs::create_dir_all(store_path.as_ref()).unwrap();

Replacement in src/lib.rs at line 366 [5.10326]

B:BD[4.12451] → [2.381:452]

        let p = extract::download_extract_deps(&index, self, pkg, &[])

[4.12451]

[4.12517]

        let p = extract::download_extract_deps(&index, self, pkg, &[], DummySink {})

Replacement in src/lib.rs at line 607 [5.10326]

∅:D[4.14073] → [5.18409:18510]

B:BD[5.18409] → [5.18409:18510]

    async fn download_url<'a>(&self, url: &str, sha256: &str) -> Result<(Downloaded, bool), Error> {

[4.14073]

[5.18510]

    async fn download_url<'a, S: futures::Sink<extract::Msg> + Unpin>(
        &self,
        url: &str,
        sha256: &str,
        mut tx: S,
    ) -> Result<(Downloaded, bool), Error>
    where
        S::Error: std::fmt::Debug,
    {

Insertion in src/lib.rs at line 624 [5.10326]

[4.14177]

        tx.send(extract::Msg::Downloading(url.to_string()))
            .await
            .unwrap();

Insertion in src/extract.rs at line 42 [5.19587]

[2.610]

[7.502]

}
/// Messages from the extractor.
#[derive(Debug)]
pub enum Msg {
    /// Downloading
    Downloading(String),
    /// Result
    Ok(Packages),
    /// Error
    Error(Error),

Replacement in src/extract.rs at line 57 [5.19587]
∅:D[6.3343] → [5.19837:19873]
∅:D[4.17671] → [5.19837:19873]
B:BD[5.19837] → [5.19837:19873]
```
pub async fn download_extract_deps(
```
[6.3343]
[5.19873]
```
pub async fn download_extract_deps<F: futures::Sink<Msg> + Clone + Send + Unpin + 'static>(
```

Replacement in src/extract.rs at line 62 [5.19587]

∅:D[2.660] → [4.17694:17725]

B:BD[5.19944] → [4.17694:17725]

B:BD[5.20068] → [4.17726:17786]

) -> Result<Packages, Error> {
    let pkg = multi_lookup(index, &package).await.unwrap();

[2.660]

[7.506]

    tx: F,
) -> Result<Packages, Error>
where
    F::Error: std::fmt::Debug,
{
    let Some(pkg) = multi_lookup(index, &package).await else {
        return Err(Error::PackageNotFound {
            pkg: package.to_string(),
        });
    };

Replacement in src/extract.rs at line 107 [5.19587]

B:BD[7.1773] → [7.1773:1837]

            let rx = spawn_extract(client.clone(), &pkg).await;

[7.1773]

[7.1837]

            let rx = spawn_extract(client.clone(), &pkg, tx.clone()).await;

Replacement in src/extract.rs at line 323 [5.19587]

B:BD[7.5478] → [7.5478:5506]

async fn spawn_extract<'a>(

[7.5478]

[7.5506]

async fn spawn_extract<'a, S: futures::Sink<Msg> + Send + Unpin + 'static>(

Replacement in src/extract.rs at line 326 [5.19587]

B:BD[7.5556] → [7.5556:5621]

) -> tokio::sync::oneshot::Receiver<Result<Downloaded, Error>> {

[7.5556]

[7.5621]

    log_tx: S,
) -> tokio::sync::oneshot::Receiver<Result<Downloaded, Error>>
where
    S::Error: std::fmt::Debug,
{

Replacement in src/extract.rs at line 337 [5.19587]

B:BD[7.5932] → [7.5932:6009]

        let (mut task, _) = match client.download_url(&url, &sha256).await {

[7.5932]

[7.6009]

        let (mut task, _) = match client.download_url(&url, &sha256, log_tx).await {

File addition: dummy_sink.rs (----------)

[5.15]

use std::convert::Infallible;
use std::pin::Pin;
#[derive(Clone, Copy)]
pub struct DummySink {}
impl<T> futures::Sink<T> for DummySink {
    type Error = Infallible;
    fn poll_ready(
        self: Pin<&mut Self>,
        _cx: &mut futures::task::Context<'_>,
    ) -> futures::task::Poll<Result<(), Infallible>> {
        futures::task::Poll::Ready(Ok(()))
    }
    fn start_send(self: Pin<&mut Self>, _item: T) -> Result<(), Infallible> {
        Ok(())
    }
    fn poll_flush(
        self: Pin<&mut Self>,
        _cx: &mut futures::task::Context<'_>,
    ) -> futures::task::Poll<Result<(), Infallible>> {
        futures::task::Poll::Ready(Ok(()))
    }
    fn poll_close(
        self: Pin<&mut Self>,
        _cx: &mut futures::task::Context<'_>,
    ) -> futures::task::Poll<Result<(), Infallible>> {
        futures::task::Poll::Ready(Ok(()))
    }
}

Insertion in elpe/lib/elpegrpc.proto at line 64 [5.55490]
[3.6581]
[4.63852]
```
    string loading = 5;
```

Replacement in elpe/lib/elpegrpc.proto at line 95 [5.55490]

B:BD[5.56055] → [5.56055:56125]

  rpc UbuntuPackage (UbuntuPackageRequest) returns (DerivationReply);

[5.56055]

[5.56125]

  rpc UbuntuPackage (UbuntuPackageRequest) returns (stream DerivationReply);

Replacement in elpe/lib/elpe.ml at line 115 [5.56160]
B:BD[4.72631] → [4.72631:72662]
```
      | `Error e -> failwith e
```
[4.72631]
[4.72662]
```
      | `Error e -> raise (DerivationError e)
```
Insertion in elpe/lib/derivation.ml at line 2 [8.445]
[8.462]
[8.462]
```
exception DerivationError of string
```

Replacement in elpe/lib/derivation.ml at line 80 [8.445]

B:BD[8.2579] → [8.2579:3246]

  Client.call ~service:"elpe.Elpe" ~rpc:"UbuntuPackage"
    ~do_request:
      (H2_lwt_unix.Client.request connection ~error_handler:(fun _ ->
           failwith "Error"))
    ~handler:
      (Client.Rpc.unary enc ~f:(fun decoder ->
           let+ decoder = decoder in
           match decoder with
           | Some decoder -> (
               Reader.create decoder |> decode |> function
               | Ok v -> v
               | Error e ->
                   failwith
                     (Printf.sprintf "Could not decode request: %s"
                        (Result.show_error e)))
           | None -> Elpegrpc.Elpe.Elpe.Derivation.Response.make ()))
    ()

[8.2579]

[8.3246]

  let result = ref None in
  let* x =
    Client.call ~service:"elpe.Elpe" ~rpc:"UbuntuPackage"
      ~do_request:
        (H2_lwt_unix.Client.request connection ~error_handler:(fun _ ->
             failwith "Error"))
      ~handler:
        (Client.Rpc.server_streaming enc ~f:(fun responses ->
             Lwt_stream.iter_s
               (fun str ->
                 Reader.create str |> decode |> function
                 | Ok (`Ok v) ->
                     result := Some v;
                     Lwt.return ()
                 | Ok (`Loading v) ->
                     Printf.eprintf "Downloading %s\n" v;
                     Stdlib.flush Stdlib.stderr;
                     Lwt.return ()
                 | Error e ->
                     failwith
                       (Printf.sprintf "Could not decode request: %s"
                          (Result.show_error e))
                 | _ -> Lwt.return ())
               responses))
      ()
  in
  match x with
  | Ok _ -> Lwt.return (Ok (Option.get !result))
  | Error e -> Lwt.return (Error e)
type arch = Amd64 | Aarch64

Replacement in elpe/lib/derivation.ml at line 112 [8.445]
B:BD[8.3247] → [8.3247:3287]
```
let ubuntu ?(release = "plucky") name =
```
[8.3247]
[8.3287]
```
let ubuntu ?(release = "plucky") ?(arch = Amd64) name =
```

Replacement in elpe/lib/derivation.ml at line 191 [8.445]

B:BD[8.5575] → [8.5575:5625]

                           Printf.printf "%s" buf

[8.5575]

[8.5625]

                           Printf.printf "%s" buf;
                           Stdlib.flush Stdlib.stdout

Replacement in elpe/lib/derivation.ml at line 201 [8.445]

B:BD[8.5944] → [8.5944:5994]

                           Printf.printf "%s" buf

[8.5944]

[8.5994]

                           Printf.eprintf "%s" buf;
                           Stdlib.flush Stdlib.stderr

Forwarding log to frontend

Dependencies

In channels

Change contents

Insertion in src/main.rs at line 9 [5.1363]

Insertion in src/main.rs at line 38 [5.1363]

Replacement in src/main.rs at line 41 [5.1363]

Replacement in src/main.rs at line 46 [5.1363]