Dapper ile Async ve Connection Yönetimi: CancellationToken, Pooling ve DI

Sürüm notu: Bu yazıdaki kodlar Dapper 2.1 ve .NET 10 içindir.

Önce Tahmin Et

Elinde baştan iptal edilmiş bir CancellationToken var ve onu QueryAsync'e veriyorsun. Kod derleniyor, IDE bir şey demiyor. Çalıştırınca ne olur: OperationCanceledException mı alırsın, yoksa row'lar mı gelir?

using var cts = new CancellationTokenSource();
cts.Cancel(); // token'ı baştan iptal ediyoruz

await using var conn = new SqliteConnection(cs);
var books = await conn.QueryAsync<Book>(
    "select id, title, price from books",
    cts.Token);

Console.WriteLine($"Rows: {books.Count()}");

Cevabı birazdan göreceğiz. Bu yazının konusu tam olarak bu satırın arkasındaki iki soru: Dapper'da async çağrıların iptali nasıl çalışır ve o conn dediğimiz connection ne zaman açılmalı, ne zaman kapanmalı, kimin elinde yaşamalı?

Hazırlık: Bu Yazının Veritabanı

Serinin geri kalanı gibi sıfır kurulumlu SQLite kullanıyoruz. Bir console projesi açıp iki paketi eklemen yeterli:

dotnet new console -n Bookstore
cd Bookstore
dotnet add package Dapper --version 2.1.89
dotnet add package Microsoft.Data.Sqlite --version 10.0.14

Örneklerin hepsi aşağıdaki küçük Setup class'ıyla başlıyor. Her çalıştırmada bookstore.db dosyasını silip authors ve books tablolarını yeniden kuruyor, böylece çıktılar her seferinde aynı oluyor:

using Dapper;
using Microsoft.Data.Sqlite;

public static class Setup
{
    public static async Task CreateAsync(string cs)
    {
        if (File.Exists("bookstore.db")) File.Delete("bookstore.db");
        await using var conn = new SqliteConnection(cs);
        await conn.ExecuteAsync("""
            create table authors (
                id      integer primary key,
                name    text not null,
                country text not null
            );
            create table books (
                id        integer primary key,
                author_id integer not null references authors(id),
                title     text not null,
                price     real not null,
                stock     integer not null check (stock >= 0)
            );
            insert into authors (id, name, country) values
                (1, 'Ursula K. Le Guin', 'USA'),
                (2, 'Frank Herbert', 'USA'),
                (3, 'Ted Chiang', 'USA');
            insert into books (id, author_id, title, price, stock) values
                (1, 1, 'A Wizard of Earthsea', 9.99, 12),
                (2, 1, 'The Left Hand of Darkness', 12.50, 4),
                (3, 2, 'Dune', 14.99, 20),
                (4, 3, 'Exhalation', 16.00, 7);
            """);
    }
}

public sealed class Book
{
    public int Id { get; set; }
    public string Title { get; set; } = "";
    public decimal Price { get; set; }
}

Program.cs tarafında da her örnek const string cs = "Data Source=bookstore.db"; ve await Setup.CreateAsync(cs); satırlarıyla açılıyor; yer kazanmak için kod bloklarında bu iki satırı tekrar etmiyorum.

Token Neden Görmezden Gelindi?

Tahmin oyununun cevabı: hiçbir exception yok, ekrana Rows: 4 basılıyor. İptal edilmiş token sessizce yutuldu.

Sebep overload çözümlemesi. QueryAsync<T> imzasında ikinci pozisyonel parametre object? param. CancellationToken bir struct, dolayısıyla box'lanıp memnuniyetle param'a oturuyor. Dapper onu parameter object'i sanıyor, property'lerini (IsCancellationRequested, CanBeCanceled, WaitHandle) SQL metninde arıyor, @IsCancellationRequested gibi bir isim bulamayınca command'a hiçbir parameter eklemiyor. Derleyici açısından her şey yolunda, çünkü bir object'i object yerine verdin.

Dapper'ın IDbConnection üzerindeki extension method'larının hiçbirinde ayrı bir CancellationToken parametresi yok. Token'ı geçirmenin tek resmi yolu CommandDefinition:

var command = new CommandDefinition(
    "select id, title, price from books",
    cancellationToken: cts.Token);
var again = await conn.QueryAsync<Book>(command);

Rows: 4
TaskCanceledException: A task was canceled.

İlk satır tahmin örneğinin çıktısı, ikinci satır CommandDefinition ile yazılmış hâlin. CommandDefinition bir struct; SQL metnini, parameter'ları, transaction'ı, timeout'u, CommandType'ı, CommandFlags'i ve token'ı tek pakette taşır. QueryAsync, ExecuteAsync, ExecuteScalarAsync, QueryFirstOrDefaultAsync ve diğerlerinin hepsinin CommandDefinition alan bir overload'u var. Bir de şunu not et: QueryAsync'te sync Query'deki gibi bir buffered parametresi yok; buffering de CommandDefinition.Flags üzerinden ayarlanıyor. Onu serinin 10. yazısında açacağız.

Pratik kural: async bir Dapper çağrısına token vereceksen hiç düşünmeden CommandDefinition yaz. Code review'da bir Dapper çağrısının argümanları arasında pozisyonel bir ct ya da cts.Token görürsen kaşın kalksın.

SQLite'ta Async Aslında Ne Kadar Async?

Token'ı doğru verdik, peki gerçekten çalışan bir sorguyu yarıda kesebilir miyiz? 20 milyon row üreten bir recursive CTE okuyalım ve 300 ms sonra iptal edelim:

using System.Diagnostics;
using Dapper;
using Microsoft.Data.Sqlite;

const string manyRowsSql = """
    with recursive n(i) as (select 1 union all select i + 1 from n where i < 20000000)
    select i from n
    """;

Console.WriteLine($"Thread before: {Environment.CurrentManagedThreadId}");
using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(300));
var sw = Stopwatch.StartNew();
try
{
    await using var conn = new SqliteConnection(cs);
    var rows = await conn.QueryAsync<long>(
        new CommandDefinition(manyRowsSql, cancellationToken: cts.Token));
    Console.WriteLine($"Rows: {rows.Count()}");
}
catch (Exception ex)
{
    Console.WriteLine($"{ex.GetType().Name}: {ex.Message}");
}
Console.WriteLine($"Elapsed: {sw.ElapsedMilliseconds} ms");
Console.WriteLine($"Thread after: {Environment.CurrentManagedThreadId}");

Thread before: 1
TaskCanceledException: A task was canceled.
Elapsed: 303 ms
Thread after: 1

İptal tam zamanında geldi. Şimdi aynı şeyi tek row dönen ama hesaplaması uzun süren bir sorguyla deneyelim: 100 milyonluk CTE'nin count(*)'u.

const string slowSql = """
    with recursive n(i) as (select 1 union all select i + 1 from n where i < 100000000)
    select count(*) from n
    """;

using var cts = new CancellationTokenSource(TimeSpan.FromMilliseconds(300));
var sw = Stopwatch.StartNew();
await using var conn = new SqliteConnection(cs);
var total = await conn.ExecuteScalarAsync<long>(
    new CommandDefinition(slowSql, cancellationToken: cts.Token));
Console.WriteLine($"Total: {total}, elapsed: {sw.ElapsedMilliseconds} ms, cancelled: {cts.IsCancellationRequested}");

Total: 100000000, elapsed: 6271 ms, cancelled: True

Token 300 ms'de iptal edildi ama sorgu 6 saniyenin üzerinde çalışıp sonucu döndü. İki örnek arasındaki fark provider'ın içinde. Microsoft.Data.Sqlite'ın dokümantasyonu açıkça söylüyor: SQLite asenkron I/O desteklemiyor, async ADO.NET method'ları senkron çalışıyor. 10.0 kaynağına bakınca ExecuteReaderAsync'in token'ı yalnızca en başta kontrol edip senkron ExecuteReader'ı çağırdığını, SqliteCommand.Cancel()'ın da boş bir method olduğunu görüyorsun. Token'ın etki ettiği tek nokta row'lar arası: Dapper buffered okumada her row için ReadAsync(token) çağırıyor ve orada iptal yakalanıyor. Tek bir uzun statement ise kesilmiyor. İlk örnekte thread id'nin 1'de kalması da aynı hikâyeyi anlatıyor: hiçbir yerde gerçek bir bekleme olmadı.

Bu, SQLite'ta async yazmayı bırak demek değil. Kodun yarın SQL Server'a ya da PostgreSQL'e taşındığında aynı kalır; bu provider'lar gerçek asenkron I/O yapar ve iptali sunucuya iletir. Ama SQLite üzerinde ölçtüğün iptal davranışını genelleme.

Connection Ne Zaman Açılıyor, Ne Zaman Kapanıyor?

Dapper'ın connection'a karşı tavrı basit: kapalı bir connection verirsen çağrının başında açar, sonunda kapatır. Açık verirsen dokunmaz.

await using (var conn = new SqliteConnection(cs))
{
    Console.WriteLine($"Before: {conn.State}");
    var count = await conn.ExecuteScalarAsync<int>("select count(*) from books");
    Console.WriteLine($"Books: {count}, after: {conn.State}");
}

Before: Closed
Books: 4, after: Closed

Bu davranış tek sorguluk işlerde harika: OpenAsync yazmana bile gerek kalmıyor. Ama aynı connection üzerinden art arda birkaç command çalıştıracaksan her çağrı ayrı bir aç/kapa demek. Transaction, geçici tablo ya da SQLite'taki CreateFunction gibi oturuma bağlı şeyler bu aç/kapa arasında kaybolur. Birden fazla command tek oturumda yaşayacaksa connection'ı kendin aç; transaction'ları bir sonraki yazıda tam olarak bu yüzden açık connection üzerinde kuracağız.

Peki her sorguda yeni bir connection oluşturmak pahalı değil mi? Burada connection pooling devreye giriyor. Dispose ettiğin connection fiziksel olarak kapanmaz, pool'a geri döner; bir sonraki Open onu oradan alır. Microsoft.Data.Sqlite'ta pooling 6.0'dan beri var ve varsayılan olarak açık. Farkı görmek için 1000 kez aç-sorgula-kapat yapalım:

await MeasureAsync("Pooling=True", "Data Source=bookstore.db");
await MeasureAsync("Pooling=False", "Data Source=bookstore.db;Pooling=False");

static async Task MeasureAsync(string label, string connectionString)
{
    var sw = Stopwatch.StartNew();
    for (var i = 0; i < 1000; i++)
    {
        await using var conn = new SqliteConnection(connectionString);
        await conn.ExecuteScalarAsync<int>("select count(*) from books");
    }
    Console.WriteLine($"{label}: {sw.ElapsedMilliseconds} ms");
}

Pooling=True: 6 ms
Pooling=False: 79 ms

Benim makinemde pool'lu 1000 tur 6 ms, pool'suz 79 ms sürdü. Ağ üzerinden konuşan bir sunucu veritabanında bu fark çok daha büyür, çünkü pool'suz her açılış TCP bağlantısı, TLS el sıkışması ve kimlik doğrulama demek. Sonuç: connection'ı "kısa yaşat, sık oluştur" kafasıyla kullan, pahalı kısmı pool zaten üstleniyor.

Tek Connection'ı Paylaşmak

Pooling'i duymamış biri "madem pahalı, bir tane açıp herkes onu kullansın" diye düşünebilir. Deneyelim: tek bir açık connection, üzerinde 50 paralel sorgu.

var shared = new SqliteConnection(cs);
await shared.OpenAsync();

var tasks = Enumerable.Range(0, 50).Select(i => Task.Run(async () =>
{
    var titles = await shared.QueryAsync<string>(
        "select title from books where id = @id", new { id = i % 4 + 1 });
    return titles.Count();
}));

try
{
    var counts = await Task.WhenAll(tasks);
    Console.WriteLine($"Done: {counts.Sum()} rows");
}
catch (Exception ex)
{
    Console.WriteLine($"{ex.GetType().Name}: {ex.Message}");
}

Done: 50 rows
Done: 50 rows
ArgumentOutOfRangeException: Index was out of range. Must be non-negative and less than the size of the collection. (Parameter 'index')

Aynı programı arka arkaya çalıştırdım; 30 denemenin 23'ünde Done: 50 rows, 7'sinde yukarıdaki exception geldi. En tehlikeli hata türü bu: çoğu zaman çalışıyor, test ortamında geçiyor, yük altında patlıyor. ADO.NET connection'ları thread-safe değil; içlerinde açık command'ların listesi, reader durumu, transaction bilgisi gibi paylaşılan state var. Bir connection'ı aynı anda tek bir iş kullanır. Singleton connection, static connection ya da birden fazla request'in paylaştığı bir connection hep aynı hataya çıkan yollar.

ASP.NET Core'da Connection Factory ve DI

Gerçek bir uygulamada connection'ı kimin oluşturacağı sorusunun en temiz cevabı bir factory. Factory yalnızca connection string'i tutar, state'i yoktur; bu yüzden singleton olarak kaydedilebilir. Her repository method'u ondan kendi connection'ını alır ve işi bitince dispose eder. dotnet new web ile açtığın bir projede Program.cs şöyle görünüyor:

using System.Data.Common;
using Dapper;
using Microsoft.Data.Sqlite;

var builder = WebApplication.CreateBuilder(args);

var cs = builder.Configuration.GetConnectionString("Bookstore") ?? "Data Source=bookstore.db";
await Setup.CreateAsync(cs);

builder.Services.AddSingleton<IDbConnectionFactory>(new SqliteConnectionFactory(cs));
builder.Services.AddScoped<BookRepository>();

var app = builder.Build();

app.MapGet("/books", async (BookRepository repo, CancellationToken ct) =>
    await repo.GetAllAsync(ct));

app.MapGet("/books/{id:int}", async (int id, BookRepository repo, CancellationToken ct) =>
    await repo.GetByIdAsync(id, ct) is { } book ? Results.Ok(book) : Results.NotFound());

app.Run();

public interface IDbConnectionFactory
{
    Task<DbConnection> OpenAsync(CancellationToken ct = default);
}

public sealed class SqliteConnectionFactory(string connectionString) : IDbConnectionFactory
{
    public async Task<DbConnection> OpenAsync(CancellationToken ct = default)
    {
        var conn = new SqliteConnection(connectionString);
        await conn.OpenAsync(ct);
        return conn;
    }
}

public sealed class BookRepository(IDbConnectionFactory factory)
{
    public async Task<IReadOnlyList<Book>> GetAllAsync(CancellationToken ct)
    {
        await using var conn = await factory.OpenAsync(ct);
        var books = await conn.QueryAsync<Book>(new CommandDefinition(
            "select id, title, price from books order by id",
            cancellationToken: ct));
        return books.AsList();
    }

    public async Task<Book?> GetByIdAsync(int id, CancellationToken ct)
    {
        await using var conn = await factory.OpenAsync(ct);
        return await conn.QuerySingleOrDefaultAsync<Book>(new CommandDefinition(
            "select id, title, price from books where id = @id",
            new { id },
            cancellationToken: ct));
    }
}

$ curl -s http://127.0.0.1:5191/books/3
{"id":3,"title":"Dune","price":14.99}
$ curl -s -o /dev/null -w "%{http_code}\n" http://127.0.0.1:5191/books/99
404

Birkaç detay önemli. Minimal API'de endpoint'e CancellationToken parametresi eklediğinde ASP.NET Core ona HttpContext.RequestAborted'ı bağlar; kullanıcı sekmeyi kapatırsa token iptal olur ve bu token CommandDefinition'a kadar akar. Factory'nin IDbConnection değil DbConnection döndürmesi bilinçli: OpenAsync, DisposeAsync ve 10. yazıda göreceğimiz QueryUnbufferedAsync hep DbConnection üzerinde. AsList() de Dapper'ın küçük bir yardımcısı; buffered sonuç zaten List<T> olduğu için kopyalamadan geri verir.

Alternatif olarak connection'ın kendisini scoped kaydedebilirsin:

builder.Services.AddScoped<DbConnection>(_ => new SqliteConnection(cs));

app.MapGet("/books/count", async (DbConnection conn, CancellationToken ct) =>
    await conn.ExecuteScalarAsync<int>(new CommandDefinition(
        "select count(*) from books", cancellationToken: ct)));

İki yaklaşımın tradeoff'u şöyle. Factory'de connection'ın ömrü kodda gözle görülür, her method kendi await using satırıyla sorumluluğu taşır, background job'lar ve paralel işler için doğal çalışır; bedeli her method'da bir satır fazlası. Scoped connection'da request başına tek connection olur, birden fazla repository aynı connection'ı ve dolayısıyla aynı transaction'ı kolayca paylaşır, dispose'u DI container request sonunda yapar; bedeli connection'ın request boyunca elde tutulması ve aynı request içinde paralel Task'lar başlatırsan bir önceki bölümdeki paylaşım hatasına geri dönmen. Ben varsayılan olarak factory'yi seçiyorum, unit of work gerektiren yerlerde scoped connection'a geçiyorum.

Sık Düşülen Tuzaklar

Token'ı pozisyonel parametre olarak vermek. Gördüğün gibi derleniyor ve hiçbir şey yapmıyor. Token her zaman CommandDefinition içinden gider.

.Result ve .Wait() ile async'i senkrona çevirmek. ASP.NET Core'da bu, thread pool'daki thread'leri bloklayıp yük altında thread pool starvation'a götürür. Async bir zincire girdiysen await ile sonuna kadar git; sync bir yerdeysen Dapper'ın sync method'larını kullan, ikisini karıştırma.

Connection'ı singleton, static field ya da HttpClient gibi düşünüp uzun ömürlü tutmak. HttpClient'ın tekrar kullanılması doğru, connection'ın değil; connection için tekrar kullanım işini pool yapar.

await using yerine using yazmak ya da hiç yazmamak. Dispose edilmeyen connection pool'a dönmez; SqlClient gibi pool'u sınırlı provider'larda üst sınıra gelindiğinde yeni istekler boşa çıkacak bir connection beklemeye başlar.

SQLite'ta ölçtüğün async davranışını sunucu veritabanlarına genellemek. SQLite'ta async çağrılar senkron çalışır ve tek bir uzun statement token ile kesilmez.

EF Core'dan gelenler için bir not: DbContext connection'ı kendi içinde açıp kapatır, sen yalnızca context'in ömrünü düşünürsün. Dapper'da bu sorumluluk doğrudan sende. EF Core serisinden tanıdığın "request başına bir DbContext" kuralının Dapper'daki karşılığı "iş birimi başına bir connection".

Küçük Bir Deneme

★☆☆ Tahmin örneğindeki SQL'i select id, title, price from books where price < @maxPrice yap ve token'ı yine pozisyonel ver. Bu sefer hangi exception'ı alıyorsun ve mesaj neden token'dan hiç bahsetmiyor?

★★☆ BookRepository'ye GetByAuthorAsync(int authorId, CancellationToken ct) ekle ve /authors/{id:int}/books endpoint'ine bağla. Sonra repository'leri scoped DbConnection alacak şekilde değiştir, tek bir endpoint'te hem BookRepository'yi hem de yeni yazacağın bir AuthorRepository'yi kullan ve ikisinin aynı connection'ı aldığını conn.GetHashCode() loglayarak doğrula.

★★★ Paylaşılan connection örneğinde Task.Run yerine sıralı bir foreach ile 50 sorguyu await ederek çalıştır. Hata hâlâ çıkıyor mu? Sonucu, connection'ın neden "aynı anda tek iş" kuralına tabi olduğuyla birlikte açıkla.

Connection'ı açık tutmanın asıl haklı sebebi birden fazla command'ı tek bir bütün hâline getirmek. Bir sonraki yazıda sipariş, sipariş kalemi ve stok güncellemesini tek transaction'a koyacağız ve transaction: parametresini unutmanın SQLite'ta neden sessiz kaldığını göreceğiz. Seri, Dapper'ın yeni sürümleri çıktıkça güncellenmeye devam edecek.