pub(crate) fn complete(
        &self,
        task_token: &TaskToken,
        status: &LocalActivityExecutionResult,
    ) -> LACompleteAction {
        let mut dlock = self.dat.lock();
        if let Some(info) = dlock.outstanding_activity_tasks.remove(task_token) {
            let exec_id = ExecutingLAId {
                run_id: info.la_info.workflow_exec_info.run_id.clone(),
                seq_num: info.la_info.schedule_cmd.seq,
            };
            dlock.id_to_tt.remove(&exec_id);
            self.semaphore.add_permit();

            match status {
                LocalActivityExecutionResult::Completed(_)
                | LocalActivityExecutionResult::TimedOut(_)
                | LocalActivityExecutionResult::Cancelled { .. } => {
                    // Timeouts are included in this branch since they are not retried
                    self.complete_notify.notify_one();
                    LACompleteAction::Report(info)
                }
                LocalActivityExecutionResult::Failed(f) => {
                    if let Some(backoff_dur) = info.la_info.schedule_cmd.retry_policy.should_retry(
                        info.attempt as usize,
                        &f.failure
                            .as_ref()
                            .map(|f| format!("{:?}", f))
                            .unwrap_or_else(|| "".to_string()),
                    ) {
                        let will_use_timer =
                            backoff_dur > info.la_info.schedule_cmd.local_retry_threshold;
                        debug!(run_id = %info.la_info.workflow_exec_info.run_id,
                               seq_num = %info.la_info.schedule_cmd.seq,
                               attempt = %info.attempt,
                               will_use_timer,
                            "Local activity failed, will retry after backing off for {:?}",
                             backoff_dur
                        );
                        if will_use_timer {
                            // We want this to be reported, as the workflow will mark this
                            // failure down, then start a timer for backoff.
                            return LACompleteAction::LangDoesTimerBackoff(
                                backoff_dur.into(),
                                info,
                            );
                        }
                        // Immediately create a new task token for the to-be-retried LA
                        let tt = dlock.gen_next_token();
                        dlock.id_to_tt.insert(exec_id.clone(), tt);

                        // Send the retry request after waiting the backoff duration
                        let send_chan = self.act_req_tx.clone();
                        let jh = tokio::spawn(async move {
                            tokio::time::sleep(backoff_dur).await;

                            send_chan
                                .send(NewOrRetry::Retry {
                                    in_flight: info.la_info,
                                    attempt: info.attempt + 1,
                                })
                                .expect("Receive half of LA request channel cannot be dropped");
                        });
                        dlock.backing_off_tasks.insert(exec_id, jh);

                        LACompleteAction::WillBeRetried
                    } else {
                        LACompleteAction::Report(info)
                    }
                }
            }
        } else {
            LACompleteAction::Untracked
        }
    }
/// Mark a local activity as having completed (pass, fail, or cancelled)
Mark a local activity as having completed (pass, fail, or cancelled)
Mark a local activity as having completed ( pass fail or cancelled )
pub(crate) fn complete(
        &self,
        task_token: &TaskToken,
        status: &LocalActivityExecutionResult,
    ) -> LACompleteAction {
        let mut dlock = self.dat.lock();
        if let Some(info) = dlock.outstanding_activity_tasks.remove(task_token) {
            let exec_id = ExecutingLAId {
                run_id: info.la_info.workflow_exec_info.run_id.clone(),
                seq_num: info.la_info.schedule_cmd.seq,
            };
            dlock.id_to_tt.remove(&exec_id);
            self.semaphore.add_permit();
            match status {
                LocalActivityExecutionResult::Completed(_)
                | LocalActivityExecutionResult::TimedOut(_)
                | LocalActivityExecutionResult::Cancelled { .. } => {
                    // Timeouts are included in this branch since they are not retried
                    self.complete_notify.notify_one();
                    LACompleteAction::Report(info)
                }
                LocalActivityExecutionResult::Failed(f) => {
                    if let Some(backoff_dur) = info.la_info.schedule_cmd.retry_policy.should_retry(
                        info.attempt as usize,
                        &f.failure
                            .as_ref()
                            .map(|f| format!("{:?}", f))
                            .unwrap_or_else(|| "".to_string()),
                    ) {
                        let will_use_timer =
                            backoff_dur > info.la_info.schedule_cmd.local_retry_threshold;
                        debug!(run_id = %info.la_info.workflow_exec_info.run_id,
                               seq_num = %info.la_info.schedule_cmd.seq,
                               attempt = %info.attempt,
                               will_use_timer,
                            "Local activity failed, will retry after backing off for {:?}",
                             backoff_dur
                        );
                        if will_use_timer {
                            // We want this to be reported, as the workflow will mark this
                            // failure down, then start a timer for backoff.
                            return LACompleteAction::LangDoesTimerBackoff(
                                backoff_dur.into(),
                                info,
                            );
                        }
                        // Immediately create a new task token for the to-be-retried LA
                        let tt = dlock.gen_next_token();
                        dlock.id_to_tt.insert(exec_id.clone(), tt);
                        // Send the retry request after waiting the backoff duration
                        let send_chan = self.act_req_tx.clone();
                        let jh = tokio::spawn(async move {
                            tokio::time::sleep(backoff_dur).await;
                            send_chan
                                .send(NewOrRetry::Retry {
                                    in_flight: info.la_info,
                                    attempt: info.attempt + 1,
                                })
                                .expect("Receive half of LA request channel cannot be dropped");
                        });
                        dlock.backing_off_tasks.insert(exec_id, jh);
                        LACompleteAction::WillBeRetried
                    } else {
                        LACompleteAction::Report(info)
                    }
                }
            }
        } else {
            LACompleteAction::Untracked
        }
    }
pub ( crate ) fn complete ( & self , task_token : & TaskToken , status : & LocalActivityExecutionResult , ) -> LACompleteAction { let mut dlock = self . dat . lock ( ) ; if let Some ( info ) = dlock . outstanding_activity_tasks . remove ( task_token ) { let exec_id = ExecutingLAId { run_id : info . la_info . workflow_exec_info . run_id . clone ( ) , seq_num : info . la_info . schedule_cmd . seq , } ; dlock . id_to_tt . remove ( & exec_id ) ; self . semaphore . add_permit ( ) ; match status { LocalActivityExecutionResult :: Completed ( _ ) | LocalActivityExecutionResult :: TimedOut ( _ ) | LocalActivityExecutionResult :: Cancelled { .. } => { // Timeouts are included in this branch since they are not retried self . complete_notify . notify_one ( ) ; LACompleteAction :: Report ( info ) } LocalActivityExecutionResult :: Failed ( f ) => { if let Some ( backoff_dur ) = info . la_info . schedule_cmd . retry_policy . should_retry ( info . attempt as usize , & f . failure . as_ref ( ) . map ( | f | format ! ( " " f ) ) . unwrap_or_else ( | | " " . to_string ( ) ) , ) { let will_use_timer = backoff_dur > info . la_info . schedule_cmd . local_retry_threshold ; debug ! ( run_id info la_info workflow_exec_info run_id seq_num info la_info schedule_cmd seq attempt info attempt will_use_timer " " backoff_dur ) ; if will_use_timer { // We want this to be reported, as the workflow will mark this // failure down, then start a timer for backoff. return LACompleteAction :: LangDoesTimerBackoff ( backoff_dur . into ( ) , info , ) ; } // Immediately create a new task token for the to-be-retried LA let tt = dlock . gen_next_token ( ) ; dlock . id_to_tt . insert ( exec_id . clone ( ) , tt ) ; // Send the retry request after waiting the backoff duration let send_chan = self . act_req_tx . clone ( ) ; let jh = tokio :: spawn ( async move { tokio :: time :: sleep ( backoff_dur ) . await ; send_chan . send ( NewOrRetry :: Retry { in_flight : info . la_info , attempt : info . attempt + 1 , } ) . expect ( " " ) ; } ) ; dlock . backing_off_tasks . insert ( exec_id , jh ) ; LACompleteAction :: WillBeRetried } else { LACompleteAction :: Report ( info ) } } } } else { LACompleteAction :: Untracked } }
Mark a local activity as having completed (pass, fail, or cancelled)
Mark a local activity as having completed ( pass fail or cancelled )
&self
task_token
&TaskToken
status
&LocalActivityExecutionResult
&self
task_token
&TaskToken
status
&LocalActivityExecutionResult