Uzun Süren ve Batch Job'lar: Chunking, Checkpoint ve Deploy Sırasında Kesilmeye Dayanmak

Sürüm notu: Bu yazıdaki kodlar .NET 10, Entity Framework Core (EF Core) 10 (Npgsql.EntityFrameworkCore.PostgreSQL 10.0.3), PostgreSQL 17 ve Hangfire 1.8.25 ile derlenip denenmiştir; süreler, Docker'da çalışan bir PostgreSQL ile 10 çekirdekli bir geliştirme makinesinde ölçüldü.

Önce Tahmin Et

Bookstore yeni fiyat listesine geçiyor: 2 milyon kitabın fiyatı yüzde 8 artacak. Bunu tek bir UPDATE ile yapmak mı, 20.000'lik parçalarla yapmak mı daha hızlı? Güncelleme sürerken müşteri hizmetleri tek bir kitabın fiyatını düzeltmek isterse, iki yöntemde ne kadar bekler? Ve iş yüzde 40'tayken uygulama yeni sürüme deploy edilirse ne olur: baştan mı başlar, kalan yüzde 60'tan mı devam eder, yoksa bazı kitaplara iki kez zam mı yapılır?

Tek Transaction'ın Bedeli

Deneme için PostgreSQL'e 2 milyon satırlık bir books tablosu doldurdum; her kitabın bir fiyatı ve şu anki fiyat listesinin sürümünü tutan bir PriceListVersion kolonu var. En kısa yol EF Core'un ExecuteUpdate'i ile tek bir SQL (Structured Query Language) cümlesi:

db.Database.SetCommandTimeout(TimeSpan.FromMinutes(10));
int updated = await db.Books
    .Where(b => b.PriceListVersion < 7)
    .ExecuteUpdateAsync(s => s
        .SetProperty(b => b.Price, b => Math.Round(b.Price * 1.08m, 2))
        .SetProperty(b => b.PriceListVersion, 7));

single statement: 2000000 rows in 6.8 s
single statement: 2000000 rows in 4.7 s
single statement: 2000000 rows in 3.9 s

Üç denemede 3,9 ile 6,8 saniye arasında sürdü. Hızlı görünüyor, ama asıl bedel süre değil. Tek cümle tek transaction demek: 2 milyon satırın hepsi commit'e kadar kilitli. Güncelleme başladıktan 2 saniye sonra aynı tablodan tek bir kitabın fiyatını güncelledim:

UPDATE books SET "Price" = 19.90 WHERE "Id" = 42;
UPDATE 1
Time: 1927.121 ms (00:01.927)

Tek satırlık bir güncelleme 1,9 saniye bekledi, çünkü büyük transaction o satırı çoktan kilitlemişti. Gerçek bir tabloda bu güncelleme dakikalar sürer ve o süre boyunca bu tabloya yazan her istek bekler; API'nin (Application Programming Interface) timeout'ları dolmaya başlar. Ayrıca iş ortada bir hata alırsa ya da process ölürse her şey geri alınır: 1,9 milyon satırı güncellemiş bir transaction'ın bütün emeği gider ve rollback'in kendisi de zaman alır.

Chunk'lar ve Keyset

Alternatif, işi küçük parçalara bölmek. Her chunk kendi transaction'ında; kilitler kısa süreli, bir hata yalnız o chunk'ı geri alır:

public sealed class PriceUpdateJob(
    IServiceScopeFactory scopeFactory,
    ILogger<PriceUpdateJob> logger) : BackgroundService
{
    private const string JobName = "price-list-v7";
    private const int TargetVersion = 7;
    private const int BatchSize = 20_000;

    protected override async Task ExecuteAsync(CancellationToken stoppingToken)
    {
        long start = Stopwatch.GetTimestamp();
        long lastId = await LoadCheckpointAsync();
        logger.LogInformation("Starting {Job} from id {LastId}", JobName, lastId);

        while (!stoppingToken.IsCancellationRequested)
        {
            await using AsyncServiceScope scope = scopeFactory.CreateAsyncScope();
            var db = scope.ServiceProvider.GetRequiredService<BookstoreDbContext>();

            // bir sonraki chunk'ın üst sınırı: keyset, OFFSET değil
            long? upperId = await db.Books
                .Where(b => b.Id > lastId)
                .OrderBy(b => b.Id)
                .Skip(BatchSize - 1)
                .Select(b => (long?)b.Id)
                .FirstOrDefaultAsync(CancellationToken.None)
                ?? await db.Books.Where(b => b.Id > lastId).MaxAsync(b => (long?)b.Id, CancellationToken.None);
            if (upperId is null)
                break; // iş bitti

            // chunk ve checkpoint aynı transaction'da: ya ikisi ya hiçbiri
            await using var tx = await db.Database.BeginTransactionAsync(CancellationToken.None);
            int updated = await db.Books
                .Where(b => b.Id > lastId && b.Id <= upperId && b.PriceListVersion < TargetVersion)
                .ExecuteUpdateAsync(s => s
                    .SetProperty(b => b.Price, b => Math.Round(b.Price * 1.08m, 2))
                    .SetProperty(b => b.PriceListVersion, TargetVersion), CancellationToken.None);
            await db.JobCheckpoints
                .Where(c => c.JobName == JobName)
                .ExecuteUpdateAsync(s => s
                    .SetProperty(c => c.LastProcessedId, upperId.Value)
                    .SetProperty(c => c.UpdatedAtUtc, DateTime.UtcNow), CancellationToken.None);
            await tx.CommitAsync(CancellationToken.None);

            lastId = upperId.Value;
            if (lastId % 200_000 == 0)
                logger.LogInformation("Checkpoint {LastId}, {Updated} rows in last chunk", lastId, updated);
        }

        logger.LogInformation("{Job} stopped at id {LastId} after {Elapsed:F1} s (cancelled={Cancelled})",
            JobName, lastId, Stopwatch.GetElapsedTime(start).TotalSeconds, stoppingToken.IsCancellationRequested);
    }
}

Üç tasarım kararı var. Birincisi keyset: her chunk "id'si lastId'den büyük olan ilk 20.000 satır" diye tanımlanıyor, OFFSET ile değil. OFFSET 1900000, veritabanının 1,9 milyon satırı okuyup atlaması demek; işin sonuna doğru her chunk bir öncekinden yavaş olur. Keyset, primary key index'iyle doğrudan doğru yere atlıyor.

İkincisi idempotent bir WHERE: PriceListVersion < TargetVersion. Bir chunk iki kez çalışırsa ikinci çalışma hiçbir satırı değiştirmez; yüzde 8'lik zam iki kez uygulanmaz. Yedinci yazıdaki kural: "şu kadar artır" yazan bir job'u "şu sürüme getir" diye koşullandır.

Üçüncüsü, chunk'ın içinde CancellationToken.None kullanmak. Bu bilinçli: kapanış sinyali geldiğinde chunk'ı yarıda kesmek istemiyoruz, chunk bitip commit edildikten sonra döngünün başında durmak istiyoruz. Chunk'ı yarıda kesmek de güvenli olurdu (transaction geri alınır), ama iş boşa gider.

Checkpoint Tablosu: Kaldığı Yeri Saklamak

Döngü lastId'yi bellekte tutuyor; process yeniden başlarsa bu bilgi gider. Bu yüzden her chunk'tan sonra bir checkpoint tablosuna yazıyoruz:

public sealed class JobCheckpoint
{
    public required string JobName { get; set; }
    public long LastProcessedId { get; set; }
    public DateTime UpdatedAtUtc { get; set; }
}

// PriceUpdateJob içinde
private async Task<long> LoadCheckpointAsync()
{
    await using AsyncServiceScope scope = scopeFactory.CreateAsyncScope();
    var db = scope.ServiceProvider.GetRequiredService<BookstoreDbContext>();
    JobCheckpoint? checkpoint = await db.JobCheckpoints.FindAsync(JobName);
    if (checkpoint is null)
    {
        db.JobCheckpoints.Add(new JobCheckpoint { JobName = JobName, UpdatedAtUtc = DateTime.UtcNow });
        await db.SaveChangesAsync();
        return 0;
    }
    return checkpoint.LastProcessedId;
}

Kritik nokta, checkpoint'in chunk ile aynı transaction'da güncellenmesi. Ayrı transaction'larda olsaydı iki commit arasında ölen bir process ya güncellenmiş ama checkpoint'e yazılmamış bir chunk bırakırdı (idempotent WHERE sayesinde zararsız, sadece tekrar okunur) ya da checkpoint'e yazılmış ama güncellenmemiş bir chunk (bu, sessiz bir veri hatası: o 20.000 kitap eski fiyatta kalır). Aynı transaction, ikisinin her zaman birlikte olmasını garanti ediyor.

İşin tamamını bu şekilde çalıştırdım ve aynı sırada az önceki tek satırlık güncellemeyi tekrar denedim:

05:23:30.317 info: Bookstore.Pricing.PriceUpdateJob[0] Starting price-list-v7 from id 0
05:23:30.881 info: Bookstore.Pricing.PriceUpdateJob[0] Checkpoint 200000, 20000 rows in last chunk
05:23:31.358 info: Bookstore.Pricing.PriceUpdateJob[0] Checkpoint 400000, 20000 rows in last chunk
[...]
05:23:35.184 info: Bookstore.Pricing.PriceUpdateJob[0] Checkpoint 2000000, 20000 rows in last chunk
05:23:35.197 info: Bookstore.Pricing.PriceUpdateJob[0] price-list-v7 stopped at id 2000000 after 5.2 s (cancelled=False)

UPDATE books SET "Price" = 19.90 WHERE "Id" = 42;
UPDATE 1
Time: 4.279 ms

Tek satırlık güncelleme 1,9 saniye yerine 4 ms'de bitti; en kötü ihtimalle tek bir chunk'ın süresi kadar bekler. Toplam süre 5,2 saniye, tek cümleyle hemen hemen aynı mertebede; chunk'lar ek sorgu ve commit maliyeti getiriyor ama bu ölçekte belirleyici değil.

Deploy Sırasında Kesilmek ve Devam Etmek

Asıl deneme. Tabloyu sıfırlayıp işi başlattım ve 2,3 saniye sonra process'e SIGTERM (terminate signal) gönderdim, yani bir deploy'un yapacağını:

05:24:02.979 info: Bookstore.Pricing.PriceUpdateJob[0] Starting price-list-v7 from id 0
05:24:03.516 info: Bookstore.Pricing.PriceUpdateJob[0] Checkpoint 200000, 20000 rows in last chunk
05:24:03.967 info: Bookstore.Pricing.PriceUpdateJob[0] Checkpoint 400000, 20000 rows in last chunk
05:24:04.440 info: Bookstore.Pricing.PriceUpdateJob[0] Checkpoint 600000, 20000 rows in last chunk
05:24:04.893 info: Microsoft.Hosting.Lifetime[0] Application is shutting down...
05:24:04.927 info: Bookstore.Pricing.PriceUpdateJob[0] Checkpoint 800000, 20000 rows in last chunk
05:24:04.927 info: Bookstore.Pricing.PriceUpdateJob[0] price-list-v7 stopped at id 800000 after 2.3 s (cancelled=True)

version 6: 1200000 books
version 7: 800000 books

"Application is shutting down" geldiğinde iş bir chunk'ın ortasındaydı. Chunk'ı tamamladı, checkpoint'i 800.000'e yazdı, döngünün başında token'ın iptal edildiğini gördü ve çıktı. Veritabanında tam olarak 800.000 kitap yeni fiyatta, checkpoint 800.000'de. Sonra işi yeniden başlattım:

05:24:06.134 info: Bookstore.Pricing.PriceUpdateJob[0] Starting price-list-v7 from id 800000
05:24:06.689 info: Bookstore.Pricing.PriceUpdateJob[0] Checkpoint 1000000, 20000 rows in last chunk
[...]
05:24:08.947 info: Bookstore.Pricing.PriceUpdateJob[0] Checkpoint 2000000, 20000 rows in last chunk
05:24:08.955 info: Bookstore.Pricing.PriceUpdateJob[0] price-list-v7 stopped at id 2000000 after 3.2 s (cancelled=False)

version 7: 2000000 books

Kaldığı yerden devam etti ve her kitap tam bir kez güncellendi. Kesilme noktası bir chunk sınırında olduğu için hiçbir satır iki kez işlenmedi; olsaydı da idempotent WHERE ikinci zammı engellerdi.

ShutdownTimeout Yetmediğinde

Bu denemede bir chunk yaklaşık yarım saniye sürdü ve host'un 30 saniyelik kapanış bütçesinin içinde rahatça bitti. Chunk boyutunu seçerken ölçmen gereken tam olarak bu: tek bir chunk'ın en kötü süresi, serinin ikinci yazısında gördüğümüz ShutdownTimeout'tan ve Kubernetes'in terminationGracePeriodSeconds'ından belirgin şekilde kısa olmalı. Değilse host chunk'ı beklerken süre dolar, process öldürülür ve chunk'ın transaction'ı geri alınır. Veri tutarlı kalır (checkpoint aynı transaction'da), ama o chunk'ın emeği her deploy'da boşa gider ve kötü bir durumda iş hiç ilerleyemez.

Kural basit: işi, her an öldürülebilecekmiş gibi tasarla. Kalıcı ilerleme kaydı, kısa ve atomik adımlar, adımların tekrarına dayanıklı mantık. Böyle bir iş için deploy bir felaket değil, kısa bir duraklama.

Hangfire'da Uzun Job'lar: Sliding Invisibility

Bu işi Hangfire job'u olarak çalıştırıyorsan bir mekanizmayı daha bilmelisin. Bir worker job'u kuyruktan aldığında, job diğer worker'lar için "görünmez" olur. Hangfire.SqlServer 1.8'in varsayılanını yazdırdım: SlidingInvisibilityTimeout=00:05:00. Kayan (sliding) kısmı, worker'ın job sürdükçe bu süreyi düzenli olarak yenilemesi; Hangfire'ın 1.6.14 sürüm notu, uzun job'lar için çok büyük timeout değerleri ayarlama ihtiyacını ortadan kaldırmak için eklendiğini yazıyor. Worker yaşadığı sürece job saatlerce sürebilir; worker ölürse yenileme durur ve süre dolunca başka bir worker job'u alır.

Bunu görmek için süreyi 30 saniyeye indirip 60 saniyelik bir katalog dışa aktarma job'u başlattım, 8 saniye sonra process'i kill -9 ile öldürüp hemen yeniden başlattım:

.UseSqlServerStorage(builder.Configuration.GetConnectionString("Hangfire"), new SqlServerStorageOptions
{
    SlidingInvisibilityTimeout = TimeSpan.FromSeconds(30) // varsayılan 5 dakika
})

public sealed class CatalogExportJob(ILogger<CatalogExportJob> logger)
{
    public async Task ExportAsync(CancellationToken ct)
    {
        logger.LogInformation("Catalog export started (pid {Pid})", Environment.ProcessId);
        await Task.Delay(TimeSpan.FromSeconds(60), ct);
        logger.LogInformation("Catalog export finished (pid {Pid})", Environment.ProcessId);
    }
}

05:25:58.450 info: Bookstore.Jobs.CatalogExportJob[0] Catalog export started (pid 28490)
05:26:05.976 kill -9
05:26:34.546 info: Bookstore.Jobs.CatalogExportJob[0] Catalog export started (pid 28547)
05:26:57.506 warn: Hangfire.Server.Worker[0] Worker stop requested while processing background job '3'. It will be re-queued.

Process 26:05.976'da öldü, yeni process job'u 26:34.5'te aldı: yaklaşık 30 saniye. Varsayılan 5 dakikayla bu süre 5 dakika olurdu; job o süre boyunca hiçbir yerde çalışmaz. Ve job baştan başladı: ExportAsync kaldığı yeri bilmiyor. Hangfire job'ları da yukarıdaki checkpoint mantığına ihtiyaç duyuyor.

Graceful shutdown'da durum farklı. Yeni process'i SIGTERM ile durdurduğumda Hangfire şunu logladı: "Worker stop requested while processing background job '3'. It will be re-queued." Job'a verdiğimiz CancellationToken iptal edildi, job kuyruğa geri kondu ve bir sonraki server onu hemen alabilir; görünmezlik süresini beklemek gerekmiyor. Bu da job method'larına CancellationToken parametresi koymak için bir sebep daha.

Throughput: Batch Size ve Paralellik

Chunk'ları sırayla işlemek zorunda değilsin. Id aralıkları birbirinden bağımsızsa, C# Asenkron serisinin, PLINQ (Parallel Language Integrated Query) dahil paralelliği anlatan "C# Paralellik: Parallel.For, Parallel.ForEachAsync, PLINQ ve Task.Run Ne Zaman" yazısındaki Parallel.ForEachAsync ile birkaç chunk'ı aynı anda çalıştırabilirsin:

int dop = int.Parse(builder.Configuration["Dop"] ?? "4");
var factory = host.Services.GetRequiredService<IServiceScopeFactory>();
long maxId = await db.Books.MaxAsync(b => b.Id);
var ranges = Enumerable.Range(0, (int)Math.Ceiling(maxId / 20_000d))
    .Select(i => (From: i * 20_000L, To: (i + 1) * 20_000L));

long start = Stopwatch.GetTimestamp();
int total = 0;
await Parallel.ForEachAsync(ranges, new ParallelOptions { MaxDegreeOfParallelism = dop }, async (range, ct) =>
{
    // her paralel kol kendi DbContext'iyle: DbContext thread-safe değil
    await using AsyncServiceScope s = factory.CreateAsyncScope();
    var chunkDb = s.ServiceProvider.GetRequiredService<BookstoreDbContext>();
    int n = await chunkDb.Books
        .Where(b => b.Id > range.From && b.Id <= range.To && b.PriceListVersion < 7)
        .ExecuteUpdateAsync(u => u
            .SetProperty(b => b.Price, b => Math.Round(b.Price * 1.08m, 2))
            .SetProperty(b => b.PriceListVersion, 7), ct);
    Interlocked.Add(ref total, n);
});
Console.WriteLine($"dop={dop}: {total} rows in {Stopwatch.GetElapsedTime(start).TotalSeconds:F1} s");

Her ölçümden önce tabloyu sıfırlayıp VACUUM çalıştırdım:

dop=1: 2000000 rows in 3.1 s
dop=2: 2000000 rows in 2.3 s
dop=4: 2000000 rows in 2.1 s
dop=8: 2000000 rows in 2.2 s

Paralellik 1'den 2'ye çıkınca süre 3,1'den 2,3 saniyeye indi, 4'te 2,1 saniye, 8'de ise artık iyileşme yok: 2,2 saniye. Darboğaz uygulama değil veritabanı; aynı tabloya yazan kollar disk, WAL (Write-Ahead Log) ve kilitler için yarışıyor. Production'da bu tavan, veritabanının API'ye ayırması gereken kapasiteyle de paylaşılıyor; gecenin bir işi için bütün veritabanını yormak, o saatte sipariş veren müşteriyi yavaşlatır. Paralellik derecesini en hızlı değere değil, veritabanının kaldırabileceği yan yüke göre seç.

Dikkat edilmesi gereken iki ayrıntı. Her paralel kol kendi scope'unu ve kendi DbContext'ini açıyor; DbContext thread-safe değil. Ve paralel chunk'larla tek bir "son işlenen id" checkpoint'i yetmez, çünkü chunk'lar sırayla bitmez; ya her chunk'ın tamamlandığını ayrı bir satırla kaydet ya da burada olduğu gibi idempotent WHERE'e güvenip yeniden başlatmada bütün aralıkları tekrar dolaş.

Üretimde Dikkat

İlerlemeyi görünür kıl. Checkpoint tablosu zaten "iş nerede" sorusunun cevabı; onu bir health check'e ya da dashboard'a bağla. 2 saattir aynı checkpoint'te duran bir iş, çalışmayan bir iştir.

Chunk boyutunu ölçerek seç. Küçük chunk daha kısa kilit ve daha kısa kapanış demek, ama daha fazla sorgu ve commit. 20.000 bu tablo ve bu makine için iyi bir değerdi; senin tablonda index'ler, trigger'lar ve replication farklı bir sayı verebilir. Tek bir chunk'ın süresini logla ve p99'unu izle.

Replication gecikmesini izle. Büyük bir toplu güncelleme, read replica'ların gecikmesini dakikalarca artırabilir. Chunk'lar arasına kısa bir bekleme koymak, replica'lara yetişme fırsatı verir.

Sık Düşülen Tuzaklar

2 milyon satırı tek transaction'da güncellemek. Uzun kilitler, büyük log ve hata durumunda her şeyin geri alınması.

Chunk'ları OFFSET ile ilerletmek. Tablonun sonuna doğru her chunk daha yavaş; ayrıca işlem sırasında satır eklenip silinirse chunk'lar kayar. Keyset kullan.

Checkpoint'i ayrı transaction'da yazmak. İki commit arasında ölen process, güncellenmemiş bir chunk'ı "bitti" diye işaretleyebilir.

Göreli güncelleme ve tekrar. Price = Price * 1.08, koşulsuz yazıldığında tekrar çalışan her chunk'a ikinci kez zam yapar. Sürüm kolonu ya da hedef değerle koşullandır.

Varsayılan görünmezlik süresine güvenmek. Hangfire'da process ölürse uzun job 5 dakika hiçbir yerde çalışmaz; çalıştığında da baştan başlar.

Küçük Bir Deneme

★☆☆ BatchSize'ı 1.000, 20.000 ve 200.000 yap ve toplam süreyi ölç. Tek satırlık güncellemenin bekleme süresi chunk boyutuyla nasıl değişiyor?

★★☆ Paralel versiyona sağlam bir checkpoint ekle: her tamamlanan aralığı (JobName, RangeStart) anahtarlı bir tabloya yaz ve yeniden başlatmada yalnız eksik aralıkları işle. SIGTERM ile kesip tekrar başlatarak dene.

★★★ CatalogExportJob'u checkpoint'li hâle getir: her 1.000 kitapta ilerlemeyi bir tabloya yaz ve job'u yeniden başladığında oradan devam ettir. kill -9 ile öldürüp görünmezlik süresinden sonra job'un kaldığı yerden devam ettiğini doğrula.

Bir sonraki yazıda bu işlerin hepsini görmeyi öğreniyoruz: Hangfire Dashboard'u production'da güvenle açmak, Quartz 4 dashboard'unun neden yetkilendirmesiz başlamadığı, "son başarılı çalışma 26 saatten eski" health check'i, OpenTelemetry ile job metric'leri ve trace'leri.