排查常见 Durable Task SDK 问题

using Microsoft.DurableTask.Client.AzureManaged;
using Microsoft.DurableTask.Worker.AzureManaged;

var connectionString = "Endpoint=http://localhost:8080;Authentication=None";

var builder = Host.CreateApplicationBuilder(args);

builder.Services.AddDurableTaskWorker()
    .AddTasks(registry =>
    {
        registry.AddOrchestrator<MyOrchestrator>();
        registry.AddActivity<MyActivity>();
    })
    .UseDurableTaskScheduler(connectionString);

builder.Services.AddDurableTaskClient()
    .UseDurableTaskScheduler(connectionString);

import {
  createAzureManagedClient,
  createAzureManagedWorkerBuilder,
} from "@microsoft/durabletask-js-azuremanaged";

const connectionString = "Endpoint=http://localhost:8080;Authentication=None";

const client = createAzureManagedClient(connectionString);
const worker = createAzureManagedWorkerBuilder(connectionString)
  .addOrchestrator(myOrchestrator)
  .addActivity(myActivity)
  .build();

from durabletask.azuremanaged.client import DurableTaskSchedulerClient
from durabletask.azuremanaged.worker import DurableTaskSchedulerWorker

endpoint = "http://localhost:8080"
task_hub_name = "my-taskhub"

# Use token_credential=None for the local emulator
worker = DurableTaskSchedulerWorker(
    host_address=endpoint,
    taskhub=task_hub_name,
    token_credential=None,
    secure_channel=False,
)
client = DurableTaskSchedulerClient(
    host_address=endpoint,
    taskhub=task_hub_name,
    token_credential=None,
    secure_channel=False,
)

使用 DurableTaskSchedulerWorkerExtensions 包中的 DurableTaskSchedulerClientExtensions 和 durabletask-azuremanaged 辅助类：

import com.microsoft.durabletask.azuremanaged.DurableTaskSchedulerClientExtensions;
import com.microsoft.durabletask.azuremanaged.DurableTaskSchedulerWorkerExtensions;

String connectionString = "Endpoint=http://localhost:8080;Authentication=None";

DurableTaskClient client = DurableTaskSchedulerClientExtensions
    .createClientBuilder(connectionString)
    .build();

DurableTaskGrpcWorker worker = DurableTaskSchedulerWorkerExtensions
    .createWorkerBuilder(connectionString)
    .addOrchestration(/* ... */)
    .addActivity(/* ... */)
    .build();

检查协调器类是否在启动期间注册到工作线程。 如果使用源生成器（[DurableTask] 属性），则注册是自动的。 否则，请手动注册：

builder.Services.AddDurableTaskWorker()
    .AddTasks(registry =>
    {
        registry.AddOrchestrator<MyOrchestrator>();
        registry.AddActivity<MyActivity>();
    })
    .UseDurableTaskScheduler(connectionString);

检查协调器函数是否已添加到工作生成器：

const worker = createAzureManagedWorkerBuilder(connectionString)
  .addOrchestrator(myOrchestrator)
  .addActivity(myActivity)
  .build();

确认协调器函数已注册到工作器：

worker.add_orchestrator(my_orchestrator)
worker.add_activity(my_activity)

检查调度程序是否已注册到工作生成器：

workerBuilder.addOrchestration(
    new TaskOrchestrationFactory() {
        @Override
        public String getName() { return "MyOrchestrator"; }

        @Override
        public TaskOrchestration create() { return new MyOrchestrator(); }
    }
);

// ❌ Wrong - non-deterministic
var now = DateTime.UtcNow;
var id = Guid.NewGuid();
var data = await httpClient.GetAsync("https://example.com/api");

// ✅ Correct - deterministic
var now = context.CurrentUtcDateTime;
var id = context.NewGuid();
var data = await context.CallActivityAsync<string>("FetchData");

// ❌ Wrong - non-deterministic
const now = new Date();
const id = crypto.randomUUID();
const data = await fetch("https://example.com/api");

// ✅ Correct - deterministic
const now = ctx.currentUtcDateTime;
const id = ctx.newGuid();
const data = yield ctx.callActivity(fetchData);

# ❌ Wrong - non-deterministic
import datetime, uuid, requests
now = datetime.datetime.utcnow()
id = str(uuid.uuid4())
data = requests.get("https://example.com/api")

# ✅ Correct - deterministic
now = ctx.current_utc_datetime
id = ctx.new_uuid()
data = yield ctx.call_activity(fetch_data)

// ❌ Wrong - non-deterministic
Instant now = Instant.now();
String id = UUID.randomUUID().toString();
String data = httpClient.send(request, BodyHandlers.ofString()).body();

// ✅ Correct - deterministic
Instant now = ctx.getCurrentInstant();
String id = ctx.newUUID().toString();
String data = ctx.callActivity("FetchData", null, String.class).await();

在.NET中，活动可以通过类名或使用源生成器的 [DurableTask] 属性进行注册。 验证活动类是否包含在工作者注册中：

builder.Services.AddDurableTaskWorker()
    .AddTasks(registry =>
    {
        registry.AddActivity<SayHello>();
    })
    .UseDurableTaskScheduler(connectionString);

从编排器调用活动时，使用类名：

string result = await context.CallActivityAsync<string>(nameof(SayHello), "Tokyo");

在 JavaScript 中，通过将函数引用传递给 worker 构建器来注册活动。 函数名称必须匹配：

const sayHello = async (ctx, name) => {
  return `Hello, ${name}!`;
};

// Register with the worker
const worker = createAzureManagedWorkerBuilder(connectionString)
  .addActivity(sayHello)
  .build();

// Call from orchestrator - use the function reference
const result = yield ctx.callActivity(sayHello, "Tokyo");

在Python中，活动是通过在工作者上使用add_activity进行注册的。 函数引用必须匹配：

def say_hello(ctx, name: str) -> str:
    return f"Hello, {name}!"

# Register with the worker
worker.add_activity(say_hello)

# Call from orchestrator - use the function reference
result = yield ctx.call_activity(say_hello, input="Tokyo")

在Java中，活动注册到提供名称的组件 TaskActivityFactory。 callActivity 中的名称必须完全匹配。

workerBuilder.addActivity(
    new TaskActivityFactory() {
        @Override
        public String getName() { return "SayHello"; }

        @Override
        public TaskActivity create() {
            return ctx -> "Hello, " + ctx.getInput(String.class) + "!";
        }
    }
);

// Call from orchestrator
ctx.callActivity("SayHello", "Tokyo", String.class).await();

在 appsettings.json 或日志配置文件中：

{
  "Logging": {
    "LogLevel": {
      "Default": "Information",
      "Microsoft.DurableTask": "Debug"
    }
  }
}

使用安全重播日志在编排重播期间避免重复日志条目。

public override async Task<string> RunAsync(
    TaskOrchestrationContext context, string input)
{
    ILogger logger = context.CreateReplaySafeLogger<MyOrchestrator>();
    logger.LogInformation("Processing input: {Input}", input);
    // ...
}

在创建工作线程时设置自定义记录器。 SDK 提供 ConsoleLogger （默认）和 NoOpLogger，也可以实现 Logger 接口：

import { ConsoleLogger } from "@microsoft/durabletask-js-azuremanaged";

const worker = createAzureManagedWorkerBuilder(connectionString)
  .logger(new ConsoleLogger())
  .addOrchestrator(myOrchestrator)
  .addActivity(myActivity)
  .build();

if (!ctx.isReplaying) {
  console.log("Processing work item");
}

配置Python标准日志记录模块：

import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("durabletask")
logger.setLevel(logging.DEBUG)

若要避免在业务流程重播期间重复日志条目，请检查 ctx.is_replaying：

if not ctx.is_replaying:
    logger.info("Processing work item")

通过使用与 SLF4J 兼容的日志记录框架（例如 Logback）来配置日志记录：

<!-- logback.xml -->
<configuration>
  <logger name="com.microsoft.durabletask" level="DEBUG" />
</configuration>

若要避免在重播期间出现重复的日志条目，请检查 ctx.getIsReplaying()：

if (!ctx.getIsReplaying()) {
    logger.info("Processing work item");
}