Azure Durable Functions框架升级后PurgeInstanceHistoryAsync执行失败

问题描述 #

框架升级后，定时清理任务触发以下错误：

{"odata.error":{"code":"InvalidInput","message":{"lang":"en-US","value":"One of the request inputs is not valid"}}}

对应的清理任务代码：

public async Task CleanupTimer
(
[TimerTrigger("0 10 */6 * * *")] TimerInfo timerInfo,
[DurableClient()] IDurableClient client
)
{
        
        DateTime beforeUTC = DateTime.UtcNow - TimeSpan.FromDays(1.5);
        await Task.WhenAll
        (
        client.CleanEntityStorageAsync(true, true, CancellationToken.None),
        Task.WhenAll
        (
        client.GetInstanceStatusesAsync
        (
        new OrchestrationStatusQueryCondition()
        {
        RuntimeStatus = ClientHelper.NonRunningStatuses
        }
        )
        .Select
        (
        items => items.Where(item => item.LastUpdatedTime < beforeUTC)
        .Select(item => client.PurgeInstanceHistoryAsync(item.InstanceId))
        )
        .ToEnumerable()
        .SelectMany(items => items)
        )
        );
}

错误原因 #

1. 异步枚举处理错误：新版Durable Functions中，GetInstanceStatusesAsync返回的是IAsyncEnumerable<OrchestrationInstanceStatus>异步枚举类型。原代码直接用.Select和.ToEnumerable()同步转换，未等待异步枚举加载完成，导致后续遍历的是无效/不完整的实例数据，调用PurgeInstanceHistoryAsync时传入了非法参数，触发InvalidInput错误。
2. 并行操作冲突：原代码同时并行执行CleanEntityStorageAsync和批量实例清理操作，可能在存储层产生资源竞争，导致输入参数被判定为无效。

解决办法 #

1. 正确处理异步枚举 #

使用异步方式完整加载实例列表，确保获取到有效数据后再执行清理。可借助System.Linq.Async包的ToListAsync()方法实现异步转换。

2. 调整执行顺序避免冲突 #

将实体存储清理与实例历史清理分开执行，避免存储层并行操作引发的冲突。

修复后的代码示例 #

using System.Linq;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Azure.WebJobs;
using Microsoft.Azure.WebJobs.Extensions.DurableTask;

public async Task CleanupTimer
(
    [TimerTrigger("0 10 */6 * * *")] TimerInfo timerInfo,
    [DurableClient()] IDurableClient client
)
{
    DateTime beforeUTC = DateTime.UtcNow - TimeSpan.FromDays(1.5);
    
    // 异步获取所有非运行状态的实例
    var queryCondition = new OrchestrationStatusQueryCondition
    {
        RuntimeStatus = ClientHelper.NonRunningStatuses
    };
    var allInstances = await client.GetInstanceStatusesAsync(queryCondition).ToListAsync();
    
    // 筛选出需要清理的历史实例
    var instancesToPurge = allInstances.Where(item => item.LastUpdatedTime < beforeUTC);
    
    // 批量清理实例历史
    var purgeTasks = instancesToPurge.Select(item => client.PurgeInstanceHistoryAsync(item.InstanceId));
    await Task.WhenAll(purgeTasks);
    
    // 单独执行实体存储清理
    await client.CleanEntityStorageAsync(true, true, CancellationToken.None);
}

额外注意事项 #

• 若项目未安装System.Linq.Async包，需通过NuGet安装：Install-Package System.Linq.Async。
• 验证ClientHelper.NonRunningStatuses包含的状态值是否合法（如Completed、Failed、Terminated等），避免传入无效状态导致查询出错。
• 确认Durable Functions新版本中IDurableClient方法的参数定义是否有变更，确保参数传递符合要求。

内容的提问来源于stack exchange，提问作者user15682658