20. Практикум: восемь задач на async/await¶

using System.Runtime.CompilerServices;

// Восемь задач. Тела методов пустые (NotImplementedException): напишите их сами. Эталонные решения — в final/. Описание задач и разбор — в README главы.
static class Practice
{
    /// <summary>
    /// Задача 1. Получить данные по всем ключам <b>параллельно</b>. Порядок результатов — как порядок ключей.
    /// Если любой запрос упал — исключение этого запроса.
    /// </summary>
    public static Task<string[]> FetchAllAsync(FakeApi api, IReadOnlyList<string> keys, CancellationToken ct)
    {
        // TODO: ваше решение
        throw new NotImplementedException();
    }

    /// <summary>
    /// Задача 2. То же, но одновременно выполняется не больше <paramref name="maxParallel"/> запросов.
    /// Порядок результатов — как порядок ключей. При отмене новые запросы не запускаются.
    /// </summary>
    public static async Task<string[]> FetchLimitedAsync(FakeApi api, IReadOnlyList<string> keys, int maxParallel, CancellationToken ct)
    {
        // TODO: ваше решение
        throw new NotImplementedException();
    }

    /// <summary>
    /// Задача 3. Запрос с таймаутом. Не уложились — <see cref="TimeoutException"/>, и сам запрос должен быть отменён.
    /// Если отменил вызывающий (<paramref name="ct"/>) — <see cref="OperationCanceledException"/>, а не TimeoutException.
    /// </summary>
    public static async Task<string> GetWithTimeoutAsync(FakeApi api, string key, TimeSpan timeout, CancellationToken ct)
    {
        // TODO: ваше решение
        throw new NotImplementedException();
    }

    /// <summary>
    /// Задача 4. Повтор при <see cref="HttpRequestException"/>: до <paramref name="maxAttempts"/> обращений,
    /// между попытками пауза baseDelay, 2×baseDelay, 4×baseDelay… Последняя ошибка пробрасывается.
    /// Отмена прерывает и запрос, и паузу и повторов не вызывает.
    /// </summary>
    public static async Task<string> GetWithRetryAsync(FakeApi api, string key, int maxAttempts, TimeSpan baseDelay, CancellationToken ct)
    {
        // TODO: ваше решение
        throw new NotImplementedException();
    }

    /// <summary>
    /// Задача 5. Одновременно спросить все зеркала и вернуть ответ первого <b>успешного</b>. Остальные запросы отменить.
    /// Если упали все — <see cref="AggregateException"/> со всеми ошибками.
    /// </summary>
    public static async Task<string> FirstSuccessAsync(IReadOnlyList<FakeApi> mirrors, string key, CancellationToken ct)
    {
        // TODO: ваше решение
        throw new NotImplementedException();
    }

    /// <summary>
    /// Задача 6. Асинхронный кеш. Одновременные запросы одного ключа вызывают фабрику <b>один раз</b>;
    /// готовое значение хранится. Упавшая загрузка в кеше не остаётся: следующий запрос пробует заново.
    /// </summary>
    public sealed class AsyncCache(Func<string, Task<string>> factory)
    {
        public Task<string> GetAsync(string key)
        {
            // TODO: ваше решение (factory(key) вызывается в вашем коде)
            throw new NotImplementedException();
        }
    }

    /// <summary>
    /// Задача 7. Лента как <see cref="IAsyncEnumerable{T}"/>: страницы читаются по мере надобности (страницу N+1 не запрашивать,
    /// пока потребитель не дошёл до конца страницы N). Отмена через WithCancellation должна прерывать чтение.
    /// </summary>
    public static async IAsyncEnumerable<string> ReadFeedAsync(FakeFeed feed, [EnumeratorCancellation] CancellationToken ct = default)
    {
        // TODO: ваше решение (в теле метода должен быть yield return)
        throw new NotImplementedException();
#pragma warning disable CS0162
        yield break;   // заглушка, чтобы метод компилировался; удалите вместе с pragma
#pragma warning restore CS0162
    }

    /// <summary>
    /// Задача 8. Конвейер «ключи → API → база»: <paramref name="workers"/> параллельных запросов к API,
    /// запись в базу одним писателем пакетами по <paramref name="batchSize"/> (последний пакет может быть меньше).
    /// Очереди между стадиями ограничены (обратное давление). Ошибка любой стадии отменяет остальные и
    /// пробрасывается наружу (настоящая причина, а не OperationCanceledException). Возвращает число записанных строк.
    /// </summary>
    public static async Task<int> RunPipelineAsync(FakeApi api, FakeDb db, IEnumerable<string> keys, int workers, int batchSize, CancellationToken ct)
    {
        // TODO: ваше решение
        throw new NotImplementedException();
    }
}

using System.Collections.Concurrent;
using System.Runtime.CompilerServices;
using System.Runtime.ExceptionServices;
using System.Threading.Channels;

// Восемь задач. В start/ тела методов пустые (NotImplementedException): напишите их сами.
// В final/ лежат эталонные решения. Описание задач и разбор — в README главы.
static class Practice
{
    /// <summary>
    /// Задача 1. Получить данные по всем ключам <b>параллельно</b>. Порядок результатов — как порядок ключей.
    /// Если любой запрос упал — исключение этого запроса.
    /// </summary>
    public static Task<string[]> FetchAllAsync(FakeApi api, IReadOnlyList<string> keys, CancellationToken ct)
    {
        // <решение>
        return Task.WhenAll(keys.Select(k => api.GetAsync(k, ct)));
        // </решение>
    }

    /// <summary>
    /// Задача 2. То же, но одновременно выполняется не больше <paramref name="maxParallel"/> запросов.
    /// Порядок результатов — как порядок ключей. При отмене новые запросы не запускаются.
    /// </summary>
    public static async Task<string[]> FetchLimitedAsync(FakeApi api, IReadOnlyList<string> keys, int maxParallel, CancellationToken ct)
    {
        // <решение>
        var results = new string[keys.Count];
        await Parallel.ForEachAsync(
            Enumerable.Range(0, keys.Count),
            new ParallelOptions { MaxDegreeOfParallelism = maxParallel, CancellationToken = ct },
            async (i, token) => results[i] = await api.GetAsync(keys[i], token));
        return results;
        // </решение>
    }

    /// <summary>
    /// Задача 3. Запрос с таймаутом. Не уложились — <see cref="TimeoutException"/>, и сам запрос должен быть отменён.
    /// Если отменил вызывающий (<paramref name="ct"/>) — <see cref="OperationCanceledException"/>, а не TimeoutException.
    /// </summary>
    public static async Task<string> GetWithTimeoutAsync(FakeApi api, string key, TimeSpan timeout, CancellationToken ct)
    {
        // <решение>
        using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
        cts.CancelAfter(timeout);
        try
        {
            return await api.GetAsync(key, cts.Token);
        }
        catch (OperationCanceledException) when (!ct.IsCancellationRequested)
        {
            throw new TimeoutException($"{key}: не уложились в {timeout.TotalMilliseconds:F0} мс");
        }
        // </решение>
    }

    /// <summary>
    /// Задача 4. Повтор при <see cref="HttpRequestException"/>: до <paramref name="maxAttempts"/> обращений,
    /// между попытками пауза baseDelay, 2×baseDelay, 4×baseDelay… Последняя ошибка пробрасывается.
    /// Отмена прерывает и запрос, и паузу и повторов не вызывает.
    /// </summary>
    public static async Task<string> GetWithRetryAsync(FakeApi api, string key, int maxAttempts, TimeSpan baseDelay, CancellationToken ct)
    {
        // <решение>
        for (int attempt = 1; ; attempt++)
        {
            try
            {
                return await api.GetAsync(key, ct);
            }
            catch (HttpRequestException) when (attempt < maxAttempts)
            {
                await Task.Delay(baseDelay * (1 << (attempt - 1)), ct);
            }
        }
        // </решение>
    }

    /// <summary>
    /// Задача 5. Одновременно спросить все зеркала и вернуть ответ первого <b>успешного</b>. Остальные запросы отменить.
    /// Если упали все — <see cref="AggregateException"/> со всеми ошибками.
    /// </summary>
    public static async Task<string> FirstSuccessAsync(IReadOnlyList<FakeApi> mirrors, string key, CancellationToken ct)
    {
        // <решение>
        using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
        var pending = mirrors.Select(m => m.GetAsync(key, cts.Token)).ToList();
        var errors = new List<Exception>();
        while (pending.Count > 0)
        {
            Task<string> done = await Task.WhenAny(pending);
            pending.Remove(done);
            if (done.IsCompletedSuccessfully)
            {
                cts.Cancel();
                return done.Result;
            }
            errors.Add(done.Exception!.InnerException ?? done.Exception);
        }
        ct.ThrowIfCancellationRequested();
        throw new AggregateException(errors);
        // </решение>
    }

    /// <summary>
    /// Задача 6. Асинхронный кеш. Одновременные запросы одного ключа вызывают фабрику <b>один раз</b>;
    /// готовое значение хранится. Упавшая загрузка в кеше не остаётся: следующий запрос пробует заново.
    /// </summary>
    public sealed class AsyncCache(Func<string, Task<string>> factory)
    {
        // <решение>
        private readonly ConcurrentDictionary<string, Lazy<Task<string>>> _items = new();

        public Task<string> GetAsync(string key)
        {
            var lazy = _items.GetOrAdd(key, k =>
            {
                Lazy<Task<string>>? self = null;
                self = new Lazy<Task<string>>(() => LoadAsync(k, self!));
                return self;
            });
            return lazy.Value;
        }

        private async Task<string> LoadAsync(string key, Lazy<Task<string>> self)
        {
            try
            {
                return await factory(key);
            }
            catch
            {
                _items.TryRemove(new KeyValuePair<string, Lazy<Task<string>>>(key, self));   // только свою запись
                throw;
            }
        }
        // </решение>
    }

    /// <summary>
    /// Задача 7. Лента как <see cref="IAsyncEnumerable{T}"/>: страницы читаются по мере надобности (страницу N+1 не запрашивать,
    /// пока потребитель не дошёл до конца страницы N). Отмена через WithCancellation должна прерывать чтение.
    /// </summary>
    public static async IAsyncEnumerable<string> ReadFeedAsync(FakeFeed feed, [EnumeratorCancellation] CancellationToken ct = default)
    {
        // <решение>
        for (int page = 0; ; page++)
        {
            Page p = await feed.GetPageAsync(page, ct);
            foreach (var item in p.Items) yield return item;
            if (!p.HasNext) yield break;
        }
        // </решение>
    }

    /// <summary>
    /// Задача 8. Конвейер «ключи → API → база»: <paramref name="workers"/> параллельных запросов к API,
    /// запись в базу одним писателем пакетами по <paramref name="batchSize"/> (последний пакет может быть меньше).
    /// Очереди между стадиями ограничены (обратное давление). Ошибка любой стадии отменяет остальные и
    /// пробрасывается наружу (настоящая причина, а не OperationCanceledException). Возвращает число записанных строк.
    /// </summary>
    public static async Task<int> RunPipelineAsync(FakeApi api, FakeDb db, IEnumerable<string> keys, int workers, int batchSize, CancellationToken ct)
    {
        // <решение>
        using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
        var input = Channel.CreateBounded<string>(20);
        var fetched = Channel.CreateBounded<string>(20);

        async Task Guard(Func<Task> stage)
        {
            try { await stage(); }
            catch { cts.Cancel(); throw; }      // ошибка одной стадии останавливает остальные
        }

        var producer = Guard(async () =>
        {
            foreach (var key in keys) await input.Writer.WriteAsync(key, cts.Token);
            input.Writer.Complete();
        });

        var fetchers = Enumerable.Range(0, workers).Select(_ => Guard(async () =>
        {
            await foreach (var key in input.Reader.ReadAllAsync(cts.Token))
                await fetched.Writer.WriteAsync(await api.GetAsync(key, cts.Token), cts.Token);
        })).ToArray();

        var closer = Guard(async () =>
        {
            await Task.WhenAll(fetchers);
            fetched.Writer.Complete();
        });

        int total = 0;
        var writer = Guard(async () =>
        {
            var batch = new List<string>(batchSize);
            await foreach (var row in fetched.Reader.ReadAllAsync(cts.Token))
            {
                batch.Add(row);
                if (batch.Count < batchSize) continue;
                await db.InsertBatchAsync(batch.ToArray(), cts.Token);
                total += batch.Count;
                batch.Clear();
            }
            if (batch.Count > 0)
            {
                await db.InsertBatchAsync(batch.ToArray(), cts.Token);
                total += batch.Count;
            }
        });

        var all = Task.WhenAll(producer, closer, writer);
        try
        {
            await all;
        }
        catch
        {
            // WhenAll бросит первое по порядку исключение, а оно может быть OperationCanceledException от «остановленной» стадии.
            var real = all.Exception!.InnerExceptions.FirstOrDefault(e => e is not OperationCanceledException) ?? all.Exception.InnerExceptions[0];
            ExceptionDispatchInfo.Capture(real).Throw();
        }
        return total;
        // </решение>
    }
}

using System.Collections.Concurrent;

// «Внешний мир» для задач: ни сети, ни базы не нужно, но задержки, сбои и отмена ведут себя как настоящие.
// Эти файлы трогать не надо: решайте в Practice.cs.

/// <summary>Условный HTTP-сервис. Считает вызовы, одновременные вызовы и отменённые вызовы.</summary>
sealed class FakeApi(int latencyMs = 100)
{
    private int _inFlight, _peak, _calls, _cancelled;
    private readonly ConcurrentDictionary<string, int> _attempts = new();

    public int LatencyMs { get; set; } = latencyMs;
    /// <summary>Первые N обращений к каждому ключу падают с HttpRequestException (503).</summary>
    public int FailFirstAttempts { get; set; }
    public bool AlwaysFail { get; set; }
    public HashSet<string> BadKeys { get; } = [];

    public int Calls => Volatile.Read(ref _calls);
    public int Peak => Volatile.Read(ref _peak);
    public int Cancelled => Volatile.Read(ref _cancelled);

    public async Task<string> GetAsync(string key, CancellationToken ct = default)
    {
        Interlocked.Increment(ref _calls);
        int now = Interlocked.Increment(ref _inFlight);
        int seen;
        while (now > (seen = Volatile.Read(ref _peak)) && Interlocked.CompareExchange(ref _peak, now, seen) != seen) { }
        try
        {
            await Task.Delay(LatencyMs, ct);
            int attempt = _attempts.AddOrUpdate(key, 1, (_, n) => n + 1);
            if (AlwaysFail || BadKeys.Contains(key) || attempt <= FailFirstAttempts)
                throw new HttpRequestException($"503 для {key} (попытка {attempt})");
            return $"data:{key}";
        }
        catch (OperationCanceledException)
        {
            Interlocked.Increment(ref _cancelled);
            throw;
        }
        finally
        {
            Interlocked.Decrement(ref _inFlight);
        }
    }
}

/// <summary>Условная база данных: пакетная вставка. Запоминает всё записанное.</summary>
sealed class FakeDb
{
    private int _writing, _peakWriting;
    public ConcurrentBag<string> Rows { get; } = [];
    public ConcurrentQueue<int> BatchSizes { get; } = [];
    public int PeakConcurrentWrites => Volatile.Read(ref _peakWriting);

    public async Task InsertBatchAsync(IReadOnlyCollection<string> rows, CancellationToken ct = default)
    {
        int now = Interlocked.Increment(ref _writing);
        int seen;
        while (now > (seen = Volatile.Read(ref _peakWriting)) && Interlocked.CompareExchange(ref _peakWriting, now, seen) != seen) { }
        try
        {
            await Task.Delay(30, ct);
            BatchSizes.Enqueue(rows.Count);
            foreach (var r in rows) Rows.Add(r);
        }
        finally { Interlocked.Decrement(ref _writing); }
    }
}

sealed record Page(IReadOnlyList<string> Items, bool HasNext);

/// <summary>Постраничная лента: 5 страниц по 10 элементов.</summary>
sealed class FakeFeed
{
    private int _pages;
    public int PagesRequested => Volatile.Read(ref _pages);

    public async Task<Page> GetPageAsync(int page, CancellationToken ct = default)
    {
        Interlocked.Increment(ref _pages);
        await Task.Delay(50, ct);
        var items = Enumerable.Range(page * 10, 10).Select(i => $"item{i}").ToList();
        return new Page(items, page < 4);
    }
}

using System.Diagnostics;

// Проверки для самопроверки. Каждая запускает ваше решение из Practice.cs на «внешнем мире» из Fakes.cs.
static class Checks
{
    public static async Task RunAsync(string[] args)
    {
        var all = new (int n, string title, Func<Task<string?>> run)[]
        {
            (1, "Параллельный сбор", Check1), (2, "Ограничение параллелизма", Check2),
            (3, "Таймаут и отмена", Check3), (4, "Повтор с экспоненциальной паузой", Check4),
            (5, "Первый успешный из зеркал", Check5), (6, "Кеш без дублей", Check6),
            (7, "Постраничная лента как поток", Check7), (8, "Конвейер на Channel", Check8),
        };
        int passed = 0, ran = 0;
        foreach (var (n, title, run) in all.Where(t => args.Length == 0 || args.Contains(t.n.ToString())))
        {
            ran++;
            var sw = Stopwatch.StartNew();
            string? error;
            try
            {
                var task = run();
                error = await Task.WhenAny(task, Task.Delay(15_000)) == task ? await task : "не уложились в 15 секунд (зависло?)";
            }
            catch (NotImplementedException) { error = "не реализовано"; }
            catch (Exception e) { error = $"исключение {e.GetType().Name}: {e.Message}"; }
            if (error is null) passed++;
            Console.WriteLine($"{(error is null ? "PASS" : "FAIL")}  {n}. {title,-34} {sw.ElapsedMilliseconds,5} мс{(error is null ? "" : "   → " + error)}");
        }
        Console.WriteLine($"\nИтого: {passed} из {ran}");
    }

    static string[] Keys(int n) => Enumerable.Range(0, n).Select(i => $"k{i}").ToArray();
    static string[] Expected(IEnumerable<string> keys) => keys.Select(k => $"data:{k}").ToArray();

    // ---- 1 ----
    static async Task<string?> Check1()
    {
        var api = new FakeApi(100);
        var keys = Keys(10);
        var sw = Stopwatch.StartNew();
        var r = await Practice.FetchAllAsync(api, keys, CancellationToken.None);
        if (!r.SequenceEqual(Expected(keys))) return "результаты не совпали или порядок нарушен";
        if (sw.ElapsedMilliseconds > 450) return $"10 запросов по 100 мс заняли {sw.ElapsedMilliseconds} мс: они идут не параллельно";
        return null;
    }

    // ---- 2 ----
    static async Task<string?> Check2()
    {
        var api = new FakeApi(50);
        var keys = Keys(40);
        var sw = Stopwatch.StartNew();
        var r = await Practice.FetchLimitedAsync(api, keys, 5, CancellationToken.None);
        if (!r.SequenceEqual(Expected(keys))) return "результаты не совпали или порядок нарушен";
        if (api.Peak > 5) return $"одновременно шло {api.Peak} запросов, лимит 5";
        if (api.Peak < 5) return $"одновременно шло только {api.Peak}: лимит 5 не использован";
        if (sw.ElapsedMilliseconds > 1200) return $"слишком медленно: {sw.ElapsedMilliseconds} мс (ожидали около 400)";

        var api2 = new FakeApi(100);
        using var cts = new CancellationTokenSource(250);
        try { await Practice.FetchLimitedAsync(api2, Keys(100), 5, cts.Token); return "при отмене не было исключения"; }
        catch (OperationCanceledException) { }
        if (api2.Calls > 20) return $"после отмены продолжали запускаться запросы: всего {api2.Calls}";
        return null;
    }

    // ---- 3 ----
    static async Task<string?> Check3()
    {
        var slow = new FakeApi(2000);
        var sw = Stopwatch.StartNew();
        try { await Practice.GetWithTimeoutAsync(slow, "a", TimeSpan.FromMilliseconds(200), CancellationToken.None); return "таймаут не сработал"; }
        catch (TimeoutException) { }
        if (sw.ElapsedMilliseconds > 700) return $"таймаут 200 мс сработал через {sw.ElapsedMilliseconds} мс";
        await Task.Delay(100);
        if (slow.Cancelled != 1) return $"сам запрос не отменён (отменено {slow.Cancelled}): после таймаута он продолжает работать впустую";

        var fast = new FakeApi(50);
        if (await Practice.GetWithTimeoutAsync(fast, "b", TimeSpan.FromSeconds(1), CancellationToken.None) != "data:b") return "быстрый запрос вернул не то";

        using var cts = new CancellationTokenSource(100);
        try { await Practice.GetWithTimeoutAsync(new FakeApi(2000), "c", TimeSpan.FromSeconds(5), cts.Token); return "внешняя отмена не сработала"; }
        catch (TimeoutException) { return "внешняя отмена превратилась в TimeoutException: различайте причины"; }
        catch (OperationCanceledException) { }
        return null;
    }

    // ---- 4 ----
    static async Task<string?> Check4()
    {
        var api = new FakeApi(10) { FailFirstAttempts = 2 };
        var sw = Stopwatch.StartNew();
        var r = await Practice.GetWithRetryAsync(api, "x", 4, TimeSpan.FromMilliseconds(100), CancellationToken.None);
        if (r != "data:x") return "результат не тот";
        if (api.Calls != 3) return $"ожидали 3 обращения (2 сбоя и успех), было {api.Calls}";
        if (sw.ElapsedMilliseconds < 290) return $"паузы слишком короткие: {sw.ElapsedMilliseconds} мс (нужно 100 + 200 мс между попытками)";
        if (sw.ElapsedMilliseconds > 900) return $"слишком долго: {sw.ElapsedMilliseconds} мс";

        var dead = new FakeApi(10) { AlwaysFail = true };
        try { await Practice.GetWithRetryAsync(dead, "y", 3, TimeSpan.FromMilliseconds(20), CancellationToken.None); return "ошибка проглочена"; }
        catch (HttpRequestException) { }
        if (dead.Calls != 3) return $"ожидали ровно 3 попытки, было {dead.Calls}";

        var api3 = new FakeApi(10) { AlwaysFail = true };
        using var cts = new CancellationTokenSource(150);
        try { await Practice.GetWithRetryAsync(api3, "z", 10, TimeSpan.FromMilliseconds(100), cts.Token); return "при отмене не было исключения"; }
        catch (OperationCanceledException) { }
        catch (HttpRequestException) { return "отмена во время паузы превратилась в HttpRequestException или повторы не прерываются"; }
        if (api3.Calls > 3) return $"после отмены повторы продолжались: {api3.Calls} обращений";
        return null;
    }

    // ---- 5 ----
    static async Task<string?> Check5()
    {
        var m0 = new FakeApi(50) { AlwaysFail = true };
        var m1 = new FakeApi(150);
        var m2 = new FakeApi(2000);
        var sw = Stopwatch.StartNew();
        var r = await Practice.FirstSuccessAsync([m0, m1, m2], "k", CancellationToken.None);
        if (r != "data:k") return "результат не тот";
        if (sw.ElapsedMilliseconds > 500) return $"ждали {sw.ElapsedMilliseconds} мс: должен вернуться первый успешный (около 150 мс)";
        await Task.Delay(100);
        if (m2.Cancelled != 1) return "медленное зеркало не отменено: запрос продолжает работать впустую";

        try
        {
            await Practice.FirstSuccessAsync([new FakeApi(20) { AlwaysFail = true }, new FakeApi(30) { AlwaysFail = true }, new FakeApi(40) { AlwaysFail = true }], "k", CancellationToken.None);
            return "когда падают все, нужно исключение";
        }
        catch (AggregateException e) when (e.InnerExceptions.Count == 3) { }
        catch (AggregateException e) { return $"AggregateException должно содержать все 3 ошибки, а в нём {e.InnerExceptions.Count}"; }
        return null;
    }

    // ---- 6 ----
    static async Task<string?> Check6()
    {
        int calls = 0;
        var cache = new Practice.AsyncCache(async key => { Interlocked.Increment(ref calls); await Task.Delay(100); return $"v:{key}"; });
        var results = await Task.WhenAll(Enumerable.Range(0, 20).Select(_ => cache.GetAsync("a")));
        if (results.Any(x => x != "v:a")) return "значения не совпали";
        if (calls != 1) return $"20 одновременных запросов одного ключа вызвали фабрику {calls} раз (нужно 1)";
        await cache.GetAsync("a");
        if (calls != 1) return "повторный запрос готового ключа снова вызвал фабрику";
        await cache.GetAsync("b");
        if (calls != 2) return "другой ключ должен вызывать фабрику отдельно";

        int attempts = 0;
        var flaky = new Practice.AsyncCache(async key =>
        {
            int n = Interlocked.Increment(ref attempts);
            await Task.Delay(50);
            if (n == 1) throw new InvalidOperationException("первый вызов падает");
            return "ok";
        });
        try { await flaky.GetAsync("x"); return "первый вызов должен был упасть"; }
        catch (InvalidOperationException) { }
        try
        {
            if (await flaky.GetAsync("x") != "ok") return "после сбоя вернулось не то значение";
        }
        catch (InvalidOperationException) { return "упавшая загрузка осталась в кеше: следующий вызов получил ту же ошибку (нужно пробовать заново)"; }
        return null;
    }

    // ---- 7 ----
    static async Task<string?> Check7()
    {
        var feed = new FakeFeed();
        var all = new List<string>();
        await foreach (var x in Practice.ReadFeedAsync(feed)) all.Add(x);
        if (all.Count != 50 || !all.SequenceEqual(Enumerable.Range(0, 50).Select(i => $"item{i}"))) return $"ожидали 50 элементов по порядку, получено {all.Count}";
        if (feed.PagesRequested != 5) return $"ожидали 5 запросов страниц, было {feed.PagesRequested}";

        var feed2 = new FakeFeed();
        var firstFifteen = new List<string>();
        await foreach (var x in Practice.ReadFeedAsync(feed2))
        {
            firstFifteen.Add(x);
            if (firstFifteen.Count == 15) break;
        }
        await Task.Delay(150);
        if (feed2.PagesRequested > 2) return $"потребитель остановился на 15-м элементе, а страниц запрошено {feed2.PagesRequested} (нужно 2): лента читает вперёд";

        var feed3 = new FakeFeed();
        using var cts = new CancellationTokenSource(80);
        try
        {
            await foreach (var _ in Practice.ReadFeedAsync(feed3).WithCancellation(cts.Token)) { }
            return "WithCancellation(token) не прервал чтение: нужен [EnumeratorCancellation]";
        }
        catch (OperationCanceledException) { }
        return null;
    }

    // ---- 8 ----
    static async Task<string?> Check8()
    {
        var api = new FakeApi(20);
        var db = new FakeDb();
        var keys = Keys(95);
        var sw = Stopwatch.StartNew();
        int n = await Practice.RunPipelineAsync(api, db, keys, workers: 4, batchSize: 10, CancellationToken.None);
        if (n != 95) return $"вернули {n}, ожидали 95";
        if (!db.Rows.OrderBy(x => x).SequenceEqual(Expected(keys).OrderBy(x => x))) return "в базе не все записи или есть дубли";
        if (api.Peak > 4) return $"одновременно {api.Peak} запросов к API, воркеров 4";
        if (db.PeakConcurrentWrites != 1) return $"запись в базу шла в {db.PeakConcurrentWrites} потока(ов) одновременно; ожидали одного писателя";
        var sizes = db.BatchSizes.ToArray();
        if (sizes.Count(s => s == 10) != 9 || sizes.Count(s => s == 5) != 1 || sizes.Length != 10) return $"пакеты: {string.Join(",", sizes)}; ожидали девять по 10 и один из 5";
        if (sw.ElapsedMilliseconds > 3000) return $"слишком медленно: {sw.ElapsedMilliseconds} мс";

        var api2 = new FakeApi(20);
        api2.BadKeys.Add("k30");
        var db2 = new FakeDb();
        var sw2 = Stopwatch.StartNew();
        try { await Practice.RunPipelineAsync(api2, db2, Keys(95), 4, 10, CancellationToken.None); return "ошибка на одном ключе должна ломать конвейер"; }
        catch (HttpRequestException) { }
        catch (Exception e) { return $"ожидали HttpRequestException (настоящую причину), получили {e.GetType().Name}"; }
        if (sw2.ElapsedMilliseconds > 3000) return "после ошибки конвейер завершался слишком долго";
        if (api2.Calls >= 95) return "после ошибки конвейер продолжал обрабатывать все ключи: остановите остальных";
        return null;
    }
}