Dive into Code

// ======================================================================
// Q: How do you test concurrent and async code?
//
// - Tests are async Task with await (Q106). Never .Result inside a test.
// - Time: FakeTimeProvider instead of Task.Delay (timers, PeriodicTimer,
//   timeouts).
// - Controlling the order: with a TaskCompletionSource you "freeze" a fake at a
//   specific point and decide yourself when it completes, as shown in the code.
// - Race tests: run the same operation many times in parallel and check the
//   invariant. E.g. 100 concurrent reservations against a stock of 10 must give
//   exactly 10 successes. For database races: an integration test against a real
//   PostgreSQL (Testcontainers, Q109).
// - A test that "usually passes" doesn't prove there is no race. Correctness
//   comes from the design (atomic operations, constraints). Tests catch
//   regressions.
// In the demo project:
// - Channel<T> and IAsyncEnumerable:
//   Infrastructure/Messaging/InMemoryMessageBus.cs.
// - PeriodicTimer, try/catch inside the loop and FOR UPDATE SKIP LOCKED:
//   OutboxProcessor.cs.
// - Task.WhenAll: GET /api/products/MIC-
//   LED/availability?mode=parallel|sequential.
// - Interlocked: QueryCounter.cs and the fake ERP.
// - Streaming: GET /api/orders/export.
// ======================================================================

[Fact]
public async Task Concurrent_callers_share_a_single_token_refresh()
{
    var gate = new TaskCompletionSource<string>(TaskCreationOptions.RunContinuationsAsynchronously);
    var calls = 0;
    var sut = new TokenProvider(fetch: _ => { Interlocked.Increment(ref calls); return gate.Task; });

    var first  = sut.GetAsync(CancellationToken.None);
    var second = sut.GetAsync(CancellationToken.None);    // starts while the first one is "hanging"
    gate.SetResult("token-1");

    Assert.Equal("token-1", await first);
    Assert.Equal("token-1", await second);
    Assert.Equal(1, calls);                               // the fetch happened only once
}

// ======================================================================
// Q: Find the bugs in this async code (written by an AI agent).
//
// - async void: an exception crashes the process and the caller doesn't know
//   when it finished. It must be async Task (Q85).
// - Singleton with a scoped DbContext: captive dependency (Q5). Fix:
//   IDbContextFactory or a new scope.
// - .Result inside the lambda: sync over async, thread pool starvation (Q120).
// - A Dictionary written from multiple threads: a race, and it can get corrupted
//   internally (Q122).
// - The same DbContext in parallel tasks: "A second operation was started on
//   this context instance" (Q63).
// - Task.WhenAll without await: SaveChangesAsync runs before the tasks finish,
//   and the exceptions are lost.
// - No CancellationToken and no limit: 5,000 SKUs = 5,000 concurrent calls to
//   the ERP.
// - N+1: one query per SKU instead of a single Where(p => skus.Contains(p.Sku))
//   (Q65).
// - One failing SKU brings down the whole batch, and the log says "Synced"
//   without any numbers.
// ======================================================================

// Original AI-generated code (buggy)
// Registered as a singleton
public class PriceSyncService(AppDbContext db, IErpClient erp, ILogger<PriceSyncService> log)
{
    private readonly Dictionary<string, decimal> _last = new();

    public async void SyncAll(List<string> skus)
    {
        var tasks = skus.Select(async sku =>
        {
            var price = erp.GetPriceAsync(sku).Result;
            _last[sku] = price;
            var p = await db.Products.FirstAsync(x => x.Sku == sku);
            p.Price = price;
        });
        Task.WhenAll(tasks);
        await db.SaveChangesAsync();
        log.LogInformation("Synced");
    }
}

// Fixed version: IDbContextFactory, bounded parallelism, one query, cancellation
public sealed class PriceSyncService(
    IDbContextFactory<AppDbContext> dbFactory, IErpClient erp, ILogger<PriceSyncService> log)
{
    public async Task SyncAllAsync(IReadOnlyCollection<string> skus, CancellationToken ct)
    {
        var prices = new ConcurrentDictionary<string, decimal>();
        await Parallel.ForEachAsync(skus,
            new ParallelOptions { MaxDegreeOfParallelism = 4, CancellationToken = ct },
            async (sku, token) =>
            {
                try { prices[sku] = await erp.GetPriceAsync(sku, token); }
                catch (HttpRequestException ex) { log.LogWarning(ex, "Price for {Sku} failed", sku); }
            });

        var found = prices.Keys.ToList();
        await using var db = await dbFactory.CreateDbContextAsync(ct);
        var products = await db.Products.Where(p => found.Contains(p.Sku)).ToListAsync(ct);   // 1 query
        foreach (var p in products) p.Price = prices[p.Sku];
        await db.SaveChangesAsync(ct);

        log.LogInformation("Synced {Updated}/{Total} prices", products.Count, skus.Count);
    }
}

// ======================================================================
// Q: The lock works on my machine but not in production with 3 pods. Why?
//
// lock, SemaphoreSlim and ConcurrentDictionary protect only within a single
// process. With 3 pods there are 3 independent locks. The fix depends on the
// problem:
// - Data in the database: atomic UPDATE, unique constraint, optimistic
//   concurrency (Q60-Q61), FOR UPDATE SKIP LOCKED for job queues.
// - "Only one instance runs the job": PostgreSQL advisory lock, a distributed
//   lock in Redis / Blob lease, or a Kubernetes CronJob.
// - Duplicate messages: idempotency key / inbox (Q31).
// ======================================================================

// Only one pod runs the nightly price import. The lock lives as long as the connection is open.
await using var conn = await dataSource.OpenConnectionAsync(ct);
await using var tryLock = new NpgsqlCommand("SELECT pg_try_advisory_lock(42)", conn);
if (!(bool)(await tryLock.ExecuteScalarAsync(ct))!) return;      // another pod is already running it
try { await importer.RunAsync(ct); }
finally
{
    await using var unlock = new NpgsqlCommand("SELECT pg_advisory_unlock(42)", conn);
    await unlock.ExecuteNonQueryAsync(CancellationToken.None);
}

// ======================================================================
// Q: Can you use HttpContext, scoped services or AsyncLocal from other threads?
//
// - HttpContext is not thread-safe and is valid only for the duration of the
//   request. Afterwards it is reused, so a background task would read another
//   request's data. Copy what you need (userId, correlationId) into plain
//   variables.
// - Parallel tasks on the same scoped service (e.g. DbContext, Q63): no.
// - AsyncLocal<T>: a value that follows the logical flow through awaits. This is
//   how Activity.Current, logging scopes and IHttpContextAccessor work. Changes
//   made inside a child task don't flow back to the parent. Use it rarely,
//   because it is hidden state.
// - There is no async constructor. Alternatives: a static CreateAsync factory,
//   initialization in IHostedService.StartAsync, or Lazy<Task<T>>.
// ======================================================================

// ✗ After the request ends, the HttpContext belongs to another request (or has been disposed)
_ = Task.Run(async () => await audit.LogAsync(HttpContext.User.Identity!.Name, orderId));

// ✓ Copy the values and send them to a queue / outbox
var user = HttpContext.User.Identity!.Name;
await auditQueue.Writer.WriteAsync(new AuditEntry(user, orderId), ct);

// ======================================================================
// Q: WhenAll, WhenAny, WhenEach, Parallel: which do you pick?
//
// Pick the tool by scenario:
// - 2-3 independent I/O calls: Task.WhenAll.
// - Many items, I/O, with a limit on concurrent calls: Parallel.ForEachAsync
//   (MaxDegreeOfParallelism) or SemaphoreSlim.
// - CPU-bound work over a collection: Parallel.ForEach, PLINQ.
// - Results in the order they complete: Task.WhenEach (.NET 9).
// - The fastest of two, or a timeout: Task.WhenAny, WaitAsync.
// - Producer/consumer within the same process: Channel<T>.
// - Across pods / services: message broker + idempotent consumer (Q30-Q31).
// - Periodic work: BackgroundService + PeriodicTimer.
// ======================================================================

// First wins: query two warehouses, keep the fastest, cancel the other
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
var primary = warehouseA.GetStockAsync(sku, cts.Token);
var backup  = warehouseB.GetStockAsync(sku, cts.Token);
var winner = await Task.WhenAny(primary, backup);
await cts.CancelAsync();             // stops the slower call
return await winner;                 // throws if the faster one failed

// WhenEach: show results as they arrive
await foreach (var done in Task.WhenEach(warehouses.Select(w => w.GetStockAsync(sku, ct))))
    progress.Report(await done);     // already completed