async fn run(
    &self,
    request: MultiPlatformProcess,
    context: Context,
  ) -> Result<FallibleProcessResultWithPlatform, String> {
    // Retrieve capabilities for this server.
    let capabilities = self.get_capabilities().await?;
    trace!("RE capabilities: {:?}", &capabilities);

    // Construct the REv2 ExecuteRequest and related data for this execution request.
    let request = self.extract_compatible_request(&request).unwrap();
    let store = self.store.clone();
    let (action, command, execute_request) = make_execute_request(&request, self.metadata.clone())?;
    let build_id = context.build_id.clone();

    debug!("Remote execution: {}", request.description);
    debug!(
      "built REv2 request (build_id={}): action={:?}; command={:?}; execute_request={:?}",
      &build_id, action, command, execute_request
    );

    // Record the time that we started to process this request, then compute the ultimate
    // deadline for execution of this request.
    let deadline_duration = self.overall_deadline + request.timeout.unwrap_or_default();

    // Ensure the action and command are stored locally.
    let (command_digest, action_digest) = with_workunit(
      context.workunit_store.clone(),
      "ensure_action_stored_locally".to_owned(),
      WorkunitMetadata::with_level(Level::Debug),
      self.ensure_action_stored_locally(&command, &action),
      |_, md| md,
    )
    .await?;

    // Check the remote Action Cache to see if this request was already computed.
    // If so, return immediately with the result.
    let cached_response_opt = with_workunit(
      context.workunit_store.clone(),
      "check_action_cache".to_owned(),
      WorkunitMetadata::with_level(Level::Debug),
      self.check_action_cache(action_digest, &self.metadata, &context),
      |_, md| md,
    )
    .await?;
    debug!(
      "action cache response: build_id={}; digest={:?}: {:?}",
      &build_id, action_digest, cached_response_opt
    );
    if let Some(cached_response) = cached_response_opt {
      return Ok(cached_response);
    }

    // Upload the action (and related data, i.e. the embedded command and input files).
    with_workunit(
      context.workunit_store.clone(),
      "ensure_action_uploaded".to_owned(),
      WorkunitMetadata::with_level(Level::Debug),
      self.ensure_action_uploaded(&store, command_digest, action_digest, request.input_files),
      |_, md| md,
    )
    .await?;

    // Submit the execution request to the RE server for execution.
    let result_fut = self.run_execute_request(execute_request, request, &context);
    let timeout_fut = tokio::time::timeout(deadline_duration, result_fut);
    let response = with_workunit(
      context.workunit_store.clone(),
      "run_execute_request".to_owned(),
      WorkunitMetadata::with_level(Level::Debug),
      timeout_fut,
      |result, mut metadata| {
        if result.is_err() {
          metadata.level = Level::Error;
          metadata.desc = Some(format!(
            "remote execution timed out after {:?}",
            deadline_duration
          ));
        }
        metadata
      },
    )
    .await;
    match response {
      Ok(r) => r,
      Err(_) => {
        debug!(
          "remote execution for build_id={} timed out after {:?}",
          &build_id, deadline_duration
        );
        Err(format!(
          "remote execution timed out after {:?}",
          deadline_duration
        ))
      }
    }
  }
/// Run the given MultiPlatformProcess via the Remote Execution API.
Run the given MultiPlatformProcess via the Remote Execution API.
Run the given MultiPlatformProcess via the Remote Execution API .
async fn run(
    &self,
    request: MultiPlatformProcess,
    context: Context,
  ) -> Result<FallibleProcessResultWithPlatform, String> {
    // Retrieve capabilities for this server.
    let capabilities = self.get_capabilities().await?;
    trace!("RE capabilities: {:?}", &capabilities);
    // Construct the REv2 ExecuteRequest and related data for this execution request.
    let request = self.extract_compatible_request(&request).unwrap();
    let store = self.store.clone();
    let (action, command, execute_request) = make_execute_request(&request, self.metadata.clone())?;
    let build_id = context.build_id.clone();
    debug!("Remote execution: {}", request.description);
    debug!(
      "built REv2 request (build_id={}): action={:?}; command={:?}; execute_request={:?}",
      &build_id, action, command, execute_request
    );
    // Record the time that we started to process this request, then compute the ultimate
    // deadline for execution of this request.
    let deadline_duration = self.overall_deadline + request.timeout.unwrap_or_default();
    // Ensure the action and command are stored locally.
    let (command_digest, action_digest) = with_workunit(
      context.workunit_store.clone(),
      "ensure_action_stored_locally".to_owned(),
      WorkunitMetadata::with_level(Level::Debug),
      self.ensure_action_stored_locally(&command, &action),
      |_, md| md,
    )
    .await?;
    // Check the remote Action Cache to see if this request was already computed.
    // If so, return immediately with the result.
    let cached_response_opt = with_workunit(
      context.workunit_store.clone(),
      "check_action_cache".to_owned(),
      WorkunitMetadata::with_level(Level::Debug),
      self.check_action_cache(action_digest, &self.metadata, &context),
      |_, md| md,
    )
    .await?;
    debug!(
      "action cache response: build_id={}; digest={:?}: {:?}",
      &build_id, action_digest, cached_response_opt
    );
    if let Some(cached_response) = cached_response_opt {
      return Ok(cached_response);
    }
    // Upload the action (and related data, i.e. the embedded command and input files).
    with_workunit(
      context.workunit_store.clone(),
      "ensure_action_uploaded".to_owned(),
      WorkunitMetadata::with_level(Level::Debug),
      self.ensure_action_uploaded(&store, command_digest, action_digest, request.input_files),
      |_, md| md,
    )
    .await?;
    // Submit the execution request to the RE server for execution.
    let result_fut = self.run_execute_request(execute_request, request, &context);
    let timeout_fut = tokio::time::timeout(deadline_duration, result_fut);
    let response = with_workunit(
      context.workunit_store.clone(),
      "run_execute_request".to_owned(),
      WorkunitMetadata::with_level(Level::Debug),
      timeout_fut,
      |result, mut metadata| {
        if result.is_err() {
          metadata.level = Level::Error;
          metadata.desc = Some(format!(
            "remote execution timed out after {:?}",
            deadline_duration
          ));
        }
        metadata
      },
    )
    .await;
    match response {
      Ok(r) => r,
      Err(_) => {
        debug!(
          "remote execution for build_id={} timed out after {:?}",
          &build_id, deadline_duration
        );
        Err(format!(
          "remote execution timed out after {:?}",
          deadline_duration
        ))
      }
    }
  }
async fn run ( & self , request : MultiPlatformProcess , context : Context , ) -> Result < FallibleProcessResultWithPlatform , String > { // Retrieve capabilities for this server. let capabilities = self . get_capabilities ( ) . await ? ; trace ! ( " " capabilities ) ; // Construct the REv2 ExecuteRequest and related data for this execution request. let request = self . extract_compatible_request ( & request ) . unwrap ( ) ; let store = self . store . clone ( ) ; let ( action , command , execute_request ) = make_execute_request ( & request , self . metadata . clone ( ) ) ? ; let build_id = context . build_id . clone ( ) ; debug ! ( " " request description ) ; debug ! ( " " build_id action command execute_request ) ; // Record the time that we started to process this request, then compute the ultimate // deadline for execution of this request. let deadline_duration = self . overall_deadline + request . timeout . unwrap_or_default ( ) ; // Ensure the action and command are stored locally. let ( command_digest , action_digest ) = with_workunit ( context . workunit_store . clone ( ) , " " . to_owned ( ) , WorkunitMetadata :: with_level ( Level :: Debug ) , self . ensure_action_stored_locally ( & command , & action ) , | _ , md | md , ) . await ? ; // Check the remote Action Cache to see if this request was already computed. // If so, return immediately with the result. let cached_response_opt = with_workunit ( context . workunit_store . clone ( ) , " " . to_owned ( ) , WorkunitMetadata :: with_level ( Level :: Debug ) , self . check_action_cache ( action_digest , & self . metadata , & context ) , | _ , md | md , ) . await ? ; debug ! ( " " build_id action_digest cached_response_opt ) ; if let Some ( cached_response ) = cached_response_opt { return Ok ( cached_response ) ; } // Upload the action (and related data, i.e. the embedded command and input files). with_workunit ( context . workunit_store . clone ( ) , " " . to_owned ( ) , WorkunitMetadata :: with_level ( Level :: Debug ) , self . ensure_action_uploaded ( & store , command_digest , action_digest , request . input_files ) , | _ , md | md , ) . await ? ; // Submit the execution request to the RE server for execution. let result_fut = self . run_execute_request ( execute_request , request , & context ) ; let timeout_fut = tokio :: time :: timeout ( deadline_duration , result_fut ) ; let response = with_workunit ( context . workunit_store . clone ( ) , " " . to_owned ( ) , WorkunitMetadata :: with_level ( Level :: Debug ) , timeout_fut , | result , mut metadata | { if result . is_err ( ) { metadata . level = Level :: Error ; metadata . desc = Some ( format ! ( " " deadline_duration ) ) ; } metadata } , ) . await ; match response { Ok ( r ) => r , Err ( _ ) => { debug ! ( " " build_id deadline_duration ) ; Err ( format ! ( " " deadline_duration ) ) } } }
Run the given MultiPlatformProcess via the Remote Execution API.
Run the given MultiPlatformProcess via the Remote Execution API .
&self
request
MultiPlatformProcess
context
Context
&self
request
MultiPlatformProcess
context
Context