Как создать бесконечную сетку DataFlow с обработкой исключений?

Исключения

В вашем init нет ничего асинхронного, это может быть стандартный синхронный конструктор. Вы можете обрабатывать исключения в своей сетке, не снимая ее, с помощью простой попытки поймать в lamda, которую вы предоставляете блоку. Затем вы можете обработать этот случай, либо отфильтровав результат вашей сетки, либо проигнорировав результат в следующих блоках. Ниже приведен пример фильтрации. В простом случае int вы можете использовать int? и отфильтровать любое значение, которое было null, или, конечно, вы можете установить любое значение магического индикатора, если хотите. Если вы действительно передаете ссылочный тип, вы можете либо вытолкнуть null, либо пометить элемент данных как грязный таким образом, чтобы это можно было проверить с помощью предиката в вашей ссылке.

public class ActionMeshProcessor {
    private TransformBlock<int, int?> Transformer { get; set; }
    private ActionBlock<int?> CompletionAnnouncer { get; set; }

    public ActionMeshProcessor(CancellationToken cancellationToken) {
        var options = new ExecutionDataflowBlockOptions {
            CancellationToken = cancellationToken,
            MaxDegreeOfParallelism = 5,
            BoundedCapacity = 5
        };


        this.Transformer = new TransformBlock<int, int?>(async input => {
            try {
                await Task.Delay(TimeSpan.FromSeconds(1)); //donig something complex here!

                if (input > 15) {
                    throw new Exception($"I can't handle this number: {input}");
                }

                return input + 1;
            } catch (Exception ex) {
                return null;
            }

        }, options);

        this.CompletionAnnouncer = new ActionBlock<int?>(async input =>
        {
            if (input == null) throw new ArgumentNullException("input");

            Console.WriteLine($"Completed: {input}");

            await Task.FromResult(0);
        }, options);

        //Filtering
        this.Transformer.LinkTo(this.CompletionAnnouncer, x => x != null);
        this.Transformer.LinkTo(DataflowBlock.NullTarget<int?>());
    }

    public async Task ProcessBlockAsync(int[] arr) {
        foreach (var item in arr) {
            await this.Transformer.SendAsync(item); // await if there are no free slots
        }
    }
}

Завершение

Вы можете выставить Complete() и Completion из своего процессора и использовать их для await завершения при завершении работы приложения, предполагая, что это единственный раз, когда вы завершаете работу меша. Кроме того, убедитесь, что вы правильно распространяете завершение по ссылкам.

    //Filtering
    this.Transformer.LinkTo(this.CompletionAnnouncer, new DataflowLinkOptions() { PropagateCompletion = true }, x => x != null);
    this.Transformer.LinkTo(DataflowBlock.NullTarget<int?>());
}        

public void Complete() {
    Transformer.Complete();
}

public Task Completion {
    get { return CompletionAnnouncer.Completion; }
}

Затем, исходя из вашего образца, наиболее вероятное место для завершения находится за пределами цикла, управляющего вашей обработкой:

public async Task ExecuteAsync(CancellationToken cancellationToken) {
    var random = new Random();

    ActionMeshProcessor processor = new ActionMeshProcessor();
    await processor.Init(cancellationToken);

    while (!cancellationToken.IsCancellationRequested) {
        await Task.Delay(TimeSpan.FromSeconds(1)); // wait before enqueuing more

        int[] items = GetItems(random.Next(3, 7));

        await processor.ProcessBlockAsync(items);
    }
    //asuming you don't intend to throw from cancellation
    processor.Complete();
    await processor.Completion();

}