在Cadence/SWF/StepFunctions中修复卡顿定时器/活动的解决方案

Question

因此，计时器在Cadence、SWF和Step函数等工作流引擎中是耐用的。耐用计时器对于需要等待很长一段时间，然后唤醒以执行一些业务逻辑的用例很有用。由于耐用性，它对各种故障具有弹性，使开发人员的编程体验和模型更好。

但是，如果要在计时器启动后对其进行更改，该怎么办？比如这个例子：

@Override
public void sampleWorkflowWithTimer(Input input){
  //...
  //some business logic before the timer
  //use a durable timer waiting for 7 days
  Workflow.sleep(Duration.ofDays(7));
  //send an email after the timer fires 
  activities.sendEmailReminder(input);
  //continue with other business 
  //...
}

计时器启动后，您可能希望计时器改为等待3天，甚至取消它。如果您简单地将工作流代码更改为此怎么办？

@Override
public void sampleWorkflowWithTimer(Input input){
  //...
  //some business logic before the timer
  //use a durable timer waiting for 73 days 
  Workflow.sleep(Duration.ofDays(3));
  //send an email after the timer fires 
  activities.sendEmailReminder(input);
  //continue with other business after the timer fires 
  //...
}

这不起作用！

在 SWF、步骤函数和节奏中，这仅适用于尚未启动计时器的新工作流执行。但在这里，您真正想要修复的是具有“卡住”计时器等待7天的工作流程。

在SWF和Cadence中，这使得事情变得更糟——“停滞”的工作流现在因为“非确定性错误”(又名NDE)而变得真正停滞，直到工作流超时。因为在幕后，Cadence/SWF把持久时间变成了工作流历史中的定时器事件，带有定时器值。在重放期间，工作流期望看到计时器以完全相同的计时器值存在。看到不同的计时器值会导致工作流中出现NDE。

同样对于活动，当我们使用非常大的超时时，它可能会卡住，但它被卡住了，因为有一个进程/线程/RPC调用被卡住，或者在部署期间主机被杀死。

那么解决办法是什么呢？

Answer 1

在步骤函数中，没有真正的解决方案。因为不可能为已经开始运行的工作流更改状态机。因此，在步骤函数中使用持久计时器（等待）时必须小心，尤其是当计时器值太大而可能会倒退时。您可以将大的计时器值分解为更小的增量，并添加一些检查点，或者使用活动来模拟计时器。这样工作肯定很痛苦。

由于SWF对工作流版本控制的支持较差，所以SWF中有一个复杂的解决方案。基本思路如下：

> < li>

将功能标志添加到工作流程输入中。新启动的工作流可以使用3天作为计时器值，而已启动的工作流不会受到影响。

将Workflow.Sleep（）更改为使用Promise，并使用新的Promise来等待操作信号。处理操作信号时，改为等待新的计时器（或跳过计时器）。请注意，这是向后兼容的更改，因为等待信号不会涉及任何工作流历史事件。

向所有已启动的工作流发送操作信号。如果要发送信号的工作流太多，并且没有一个好的方法来查找和发送信号，那么在这里可能会很乏味。

这种方法也可以与Cadence一起使用，它与解决方案#3非常相似。

这个答案的其余部分描述了Cadence中可以采用的三种不同方法。

解决方案#1：重置工作流程

这可能是最容易理解和应用的解决方案。与上述示例相同，您将工作流更新为使用 3 天作为计时器值，并让一些现有工作流陷入 NDE 状态。然后收集这些工作流，并使用“上次决定已完成”重置类型来重置它们。

./cadence --do <domain> wf reset --resetType LastDecisionCompleted -w <workflowID> -r <runID> --reason "some reason"

Last决策已完成resetType意味着忘记上次工作流决策任务的结果。在这种情况下，它正是安排7天计时器的那个。

如果有许多要重置的命令，则可能需要使用批量重置命令。请参阅有关重置功能的 CLI 文档。

在幕后，reset将让停滞的工作流忘记最后7天的计时器，并在安排计时器之前继续。因为代码已更新为使用3天计时器，所以工作流现在将按预期运行。

解决方案#2:版本批量重置

Cadence拥有更强大的工作流版本支持:

“getVersion 用于安全地对工作流定义执行向后不兼容的更改。不允许在工作流运行时更新工作流代码，因为它会破坏确定性。解决方案是既有用于重播现有工作流的旧代码，也有首次执行工作流时使用的新代码。

首次执行时，将返回最大支持的版本。此版本作为标记事件记录到工作流历史记录中。即使更改了 maxSupport 版本，也会在重播时返回录制的版本。默认版本常量包含以前未进行版本控制的代码版本。

我们可以使用此版本控制来更改计时器：

@Override
public void sampleWorkflowWithTimer(Input input){
  //...
  //some business logic before the timer
  //use a durable timer waiting for 7 days
  int version = Workflow.getVersion("timerChange", Workflow.DEFAULT_VERSION, 1);
  if (version == Workflow.DEFAULT_VERSION) {
     Workflow.sleep(Duration.ofDays(7));
  } else {
   // Because the workflow has waited for some time,
   // you may want to sleep for 3-timeAlreadyElapsed instead
     Workflow.sleep(Duration.ofDays(3));
  }
  //send an email after the timer fires 
  activities.sendEmailReminder(input);
  //continue with other business 
  //...
}

请注意，在SWF中的工作流输入中使用这种功能强大的版本控制的好处是，不仅新启动的工作流将使用3天计时器值，而且已经启动的工作流也将使用3天计时器，只要它们还没有启动7天计时器。

然后，我们可以修复启动7天计时器的工作流程。我们将使用“上次决定完成”重置类型来重置这些工作流。但是，由于版本控制，重置变得稍微容易使用 -

节奏自动将搜索属性添加到使用版本控制的工作流。它允许您在历史记录中查找具有特定版本的工作流。在这种情况下，将启动 3 天计时器的工作流将具有搜索属性“节奏更改版本”，其值为“计时器更改-1”。因此，要找到卡住的工作流，我们可以使用以下 SQL：

WorkflowTYpe = “YourWorkflowType” AND CloseTime = missing AND StartTime < “NewCodeDeployTime” AND CadenceChangeVersion != “timerChange-1”

其中，WorkflowTYpe = “您的工作流类型”仅表示该特定工作流类型，关闭时间 = 缺失表示仅打开工作流，“开始时间”

上面的SQL可以包括从旧代码开始的工作流，但还没有启动计时器。如果你想更精确，你也可以在SQL中包含历史长度。你需要弄清楚卡在7天计时器上的历史长度（事件计数）的大致范围是多少。

一旦有了SQL，然后使用批量重置命令来重置工作流:

./cadence wf reset-batch --query ' WorkflowType= “YourWorkflowType” AND CloseTime = missing AND StartTime < "NewCodeDeployTime" AND CadenceChangeVersion != "timerChange-1" ' --resetType LastDecisionCompleted --reason "some reason"

解决方案#3:版本控制批处理信号

这种方法与我们为SWF描述的方法非常相似。首先，我们使用版本控制来更改工作流代码，而不是使用功能标志。

@Override
public void sampleWorkflowWithTimer(Input input){
  //...
  //some business logic before the timer
  //use a durable timer waiting for 7 days
  int version = Workflow.getVersion("timerChange", Workflow.DEFAULT_VERSION, 1);
  if (version == Workflow.DEFAULT_VERSION) {
     final boolean received = Workflow.await(Duration.ofDays(7), 
                                  () -> this.operationSignal == true);
     if(received){
        // Because the workflow has waited for some time,
        // you may want to sleep for 3-timeAlreadyElapsed instead
        Workflow.sleep(Duration.ofDays(3));
     }
  } else {
     Workflow.sleep(Duration.ofDays(3));
  }
  //send an email after the timer fires 
  activities.sendEmailReminder(input);
  //continue with other business 
  //...
}

@Override
  public void operationSignal(final String signal) {
    // you can add more cases to this operationSignal
    this.operationSignal = true;
    LOGGER.info("receive operationSignal: " + signal );
  }

就像上面说的，这个版本控制不同于SWF输入特征标志。从旧代码开始的工作流也可以使用3天计时器，只要它们还没有开始7天计时器。

值得一提的是，Workflow.Sleep（）Workflow.await（）的更改是向下兼容的。这是因为它们都安排了一个具有相同值的计时器——7天。等待信号不需要任何历史事件。

现在您可以向所有等待的工作流发送操作信号。像上面一样，我们可以使用SQL来搜索所有这些工作流。然后使用批处理信号命令给它们发信号。

./cadence wf batch start --query 'WorkflowType= “YourWorkflowType” AND CloseTime = missing AND StartTime < "NewCodeDeployTime" AND CadenceChangeVersion != "timerChange-1" ' --reason "some reason" --bt signal --input "anything"
--sig SampleWorkflow::operationSignal 

一旦卡住的工作流收到操作信号，它将从7天计时器中解封，并在“收到”逻辑中执行新代码。

请注意，在Cadence中，批量向工作流发送信号比SWF更方便。批处理作业保证在Cadence中作为系统工作流执行。

如果您计划并启动了一个超时时间很长的活动，但后来不想等待它完成/超时/失败，该怎么办？工作流现在由于等待活动而卡住。

预防

在进入解决方案之前，值得注意的是，在Cadence/SWF/Step函数中，使用没有适当心跳的长活动超时被认为是一种反模式。所以你应该从一开始就避免。如果您希望某项活动运行很长时间，例如

使用heartbeat不仅对由于某些IO/依赖性而停滞的活动很重要。这在工作者部署或活动工作者失败的情况下更可能发生，并且您需要重新启动活动。

class MyActivitiesImpl implements MyActivities {

    @Override
    public String myHeartbeatActivity() {
      ...
      // after any IO/RPC call/some time period
      Activity.heartbeat(heartbeatDetails);
      ...

请注意，对cadence服务器的实际心跳调用是由客户端SDK优化的。这样，如果你每1ms调用一次，它就不会有任何性能问题。在内部，SDK将在心跳超时的80%左右时决定进行RPC调用。

那么，如果我的活动已经卡住了怎么办？

然而，错误总是会发生，因为我们都是人。

假设我们的工作流代码如下所示。

@Override
public void sampleWorkflowWithLongTimeoutActivity(Input input){
  //...
  //some business logic before the activity
  activities.helloActivity(input);
  //continue with other business 
  //...
}

现在，许多工作流都停留在 helloActivity 上。由于此活动由于不正确的超时而卡住，因此为了缓解此问题，您应该做的第一件事是更新活动选项以使用正确的超时值，或者如果活动是长时间运行的活动，则应使用检测信号。

更新活动超时选项是Cadence和SWF中向后兼容的更改，无需任何版本控制。但是，活动超时是在步骤函数的状态机中指定的，状态机的任何微小更改仅对新工作流执行生效。因此，对于Cadence和SWF，修复超时选项将适用于尚未启动活动的任何工作流，但对于Step功能，它仅适用于从头开始的全新工作流。

使用错误超时值启动活动的工作流如何？

显然，在Step Functions状态机中，您无能为力。您可以杀死工作流并重新启动它们，但如果工作流有一些值表明您不想重新启动它们，那就很乏味了。然后，您唯一应该做的就是在活动方面：您应该仔细编写活动代码（确保没有死循环或死等待块）使用适当的指标监控活动执行。您可以在活动中添加逻辑，以便在单独的线程中早期将故障返回到状态机，如果有重试目前无济于事的情况。单独的线程会尝试在工作流方面强制执行正确的超时（但实际上并不相同）。

解决方案#1

对于SWF/Cadence，您可以对工作流代码中的停滞计时器使用类似的解决方案。

@Override
public void sampleWorkflowWithLongTimeoutActivity(Input input){
  //...
  //some business logic before the activity
  Promise<void> hello = Async.function(activities::helloActivity, input);
  Workflow.await(()-> hello.isCompleted() || this.operationSignal == true );
  if(this.operationSignal){
   // add your logic to handle the situation that we skip the wrong activity timeout.You may want to schedule the same activity again with correct timeouts
  }else{
  //continue with other business like before to be compatible 
  }
 
}

这里的诀窍是将同步更改为异步。将单个活动从同步更改为异步通常会导致 NDE，但在这种情况下，工作流会立即等待活动和信号。在内部，这将具有相同的工作流历史记录，因此它是向后兼容的更改。

解决方案2

幸运的是，如果您使用的是Cadence，只需使用Last决策完成重置类型重置工作流程，它是一个救星。

此外，如果您有太多要重置，您可以使用批量重置。

解决方案#3

您还可以使用 CLI 命令完成或失败活动：

./cadence --do <domain> wf activity complete -w <workflowID> -r <runID> --activity_id <activityID> --result <result> --identity <some_identity_string>

和

./cadence --do <domain> wf activity fail -w <workflowID> -r <runID> --activity_id <activityID> --reason <reason> --detail <detail> --identity <some_identity_string>