Forma yapıştırılan yüz bin satır: işi kuyruğa alın

Biri metin kutusuna yüz bin satır yapıştırıp düğmeye basıyor. Bir dakika kadar sonra tarayıcı gateway timeout gösteriyor. Satırların hepsi eksik değil, yarısı orada; çünkü proxy pes ettikten sonra istek sunucuda çalışmaya devam etti. Kullanıcı da gayet makul biçimde listeyi tekrar yapıştırıyor.

Gerçekte ne oluyor

Sıralama her seferinde aynı, yalnızca sayılar değişiyor:

1. Tarayıcı yaklaşık üç megabaytlık metni gönderir.
2. Proxy bunu alır ve altmış saniyelik okuma timeout'uyla uygulamaya iletir.
3. Uygulama metni ayrıştırır, her satırı normalize eder ve mevcut tabloya karşı tek tek tekrar kontrolü yapar.
4. Kırk bininci satır civarında altmış saniye biter. Proxy bağlantıyı kapatır ve 504 döner.
5. Uygulamanın istemcinin gittiğinden haberi yoktur. İşi bitene ya da süreç yeniden başlatılana kadar insert etmeyi sürdürür.
6. Kullanıcı bir hata görür ve veritabanında, kullanıcının var olmadığını sandığı kırk bin satır durur.

Satırın para ettiği bir sistemde pahalı adım altıncı adımdır. Yazılan satırlar gerçektir: kuyruktadırlar, ücretleri kesilmiş olabilir, hatta gönderilmiş olabilirler. İkinci yapıştırma aynı listeyi yeniden üretir; artık alıcılar iki kopya alır ve hesaptan iki kez para çıkar. Hata mesajı işlemin başarısız olduğunu söylediği için, olaya dahil olan hiç kimsenin aksini düşünmek için sebebi yoktur.

Listenin büyüklüğü de tek başına sorun değil. Aynı uç, otuz bin satırı kırk saniyede bitirdiği için aylarca sorunsuz çalışır; kullanıcı listesini büyüttüğü gün, kodda hiçbir şey değişmeden kırılır. Yani hatayı tetikleyen şey bir dağıtım değil, müşterinin işinin büyümesidir ve o yüzden kimse ilk anda değişiklik günlüğünde arayacak bir şey bulamaz.

İki tasarım kararı bu hatayı olası olmaktan çıkarıp kesinleştiriyor. Birincisi işi istek içinde yapmak: son teslim tarihi, işten hiçbir şey bilmeyen bir proxy'nin elindedir. İkincisi, iş biriminin sınırına karşılık gelen bir transaction olmadan yazarak ilerlemek; böylece herhangi bir yerde kesilme, kimsenin tasarlamadığı bir durum bırakır.

Nasıl görülür

Proxy log'u bu hataya bakıp bakmadığınızı hemen söyler. Upstream yanıt süresini ve durumu yazdırın, sonra sayın:

awk '$NF ~ /^[0-9.]+$/ && $NF > 30 { print $7, $9, $NF }' /var/log/nginx/access.log \
  | sort -k3 -rn | head
# /api/contacts/import 504 60.001
# /api/contacts/import 504 60.000
# /api/contacts/import 200 47.180

Tam olarak timeout değerinde duran 504'ler, kesilmiş isteklerdir. Kırk yedi saniyedeki 200 ise bugün yetişen, gelecek ay yetişmeyecek olandır.

Sonra o isteklerin geride bıraktığı duruma bakın:

SELECT list_id,
       count(*)                          AS rows,
       min(created_at)                   AS started,
       max(created_at) - min(created_at) AS spread
FROM contacts
WHERE created_at > now() - interval '7 days'
GROUP BY list_id
HAVING max(created_at) - min(created_at) > interval '20 seconds'
ORDER BY rows DESC;

Satırları bir dakikaya yayılmış bir liste, toplu işle değil istekle yazılmış bir listedir. Ayrıca satır sayısı birbirine çok yakın ve birkaç dakika arayla oluşturulmuş iki liste buluyorsanız, ikincisi o ikinci yapıştırmadır.

Çözüm

Ucu ikiye ayırın: biri kabul eder, diğeri çalışır.

Kabul eden taraf neredeyse hiçbir şey yapmaz. Boyutu doğrular, ham veriyi geldiği gibi saklar, bir iş satırı yazar ve yanıt döner:

app.post("/api/contacts/import", async (req, res) => {
  const raw = req.body.text ?? "";
  const lines = countLines(raw);

  if (lines > MAX_ROWS) {
    return res.status(413).json({
      error: "too_many_rows",
      limit: MAX_ROWS,
      submitted: lines,
      message: `Bu listede ${lines} satır var. Sınır ${MAX_ROWS}. Listeyi bölüp parça parça gönderin.`,
    });
  }

  const key = req.get("Idempotency-Key") ?? sha256(`${req.accountId}:${raw}`);
  const job = await db.jobs.upsertByKey({
    key, accountId: req.accountId, payload: raw, total: lines, state: "DRAFT",
  });

  res.status(202).json({ jobId: job.id, total: job.total, poll: `/api/jobs/${job.id}` });
});

Burada üç şey oluyor ve üçü de önemli. Ham veri, hiçbir şey yorumlanmadan önce saklanıyor; böylece iş, kullanıcıyı hiç rahatsız etmeden tekrarlanabiliyor. Idempotency anahtarı, ikinci kez gelen aynı gönderimin yeni iş değil mevcut işi dönmesini sağlıyor. Sınır ise hata yığını yerine bir sayı ve bir talimat üretiyor.

Çalışan taraf saklanan veriyi okur, parçalar hâlinde ilerler ve nereye geldiğini kaydeder:

const CHUNK = 2000;

while (job.processed < job.total) {
  const rows = parseSlice(job.payload, job.processed, CHUNK).map(normalise);

  await db.tx(async (t) => {
    await t.query(
      `INSERT INTO contacts (list_id, msisdn, name)
       SELECT $1, r.msisdn, r.name FROM unnest($2::contact_row[]) AS r
       ON CONFLICT (list_id, msisdn) DO NOTHING`,
      [job.listId, rows]
    );
    await t.query(
      `UPDATE jobs SET processed = $2, state = 'RUNNING' WHERE id = $1 AND processed = $3`,
      [job.id, job.processed + rows.length, job.processed]
    );
  });

  job.processed += rows.length;
}

Insert ile ilerleme güncellemesi aynı transaction içinde olduğu için kaydedilen konum her zaman doğrudur. Worker iki parça arasında öldürülürse, bir sonraki çalıştırma processed değerini okur ve oradan devam eder. AND processed = $3 koşulu da iki worker'ın aynı işi birlikte ilerletmesini engeller.

Tekrar eleme insert anında, (list_id, msisdn) üzerindeki unique index ile ve ham değer yerine normalize edilmiş değer üzerinde yapılır. Önce normalize edip tekilliği veritabanına bırakmak, eşzamanlılıkta ayakta kalan tek yöntemdir. Sonradan çalışan bir temizlik sorgusuyla elemek, tekrar eden satırın tabloyu tüketen her şey tarafından alınacak kadar uzun süre var olduğu anlamına gelir ve gönderim yapan bir sistemde bu, harcanmış para demektir. Aynı gerekçe kuyruğun tamamı için geçerli: sağlayıcıdan gelen geçici bir reddetme işi çöpe atmanın sebebi değildir, tekrar eden satır da sonradan onarılacak bir şey değildir.

Bunun maliyeti bir işler tablosu, bir worker ve bir ilerleme ucu, artı spinner yerine çubuk gösterebilen bir arayüzdür. Ortadan kaldırmadığı şey üst sınır ihtiyacıdır. Kuyruk, büyük bir içe aktarımı hayatta kalabilir yapar, bedava yapmaz.

Çalıştığını nasıl doğrularsınız

Kabul eden taraf, boyuttan bağımsız olarak milisaniyeler içinde yanıt vermeli. Gerçek bir dosyayla ölçün:

curl -s -o /dev/null -w "%{http_code} %{time_total}\n" \
  -H "Content-Type: application/json" \
  -H "Idempotency-Key: import-2f9c" \
  -d @/tmp/list-100k.json \
  https://example.com/api/contacts/import
# 202 0.284

Aynı dosyayı aynı anahtarla ikinci kez gönderin. Durum yine 202, iş kimliği aynı kimlik ve satır sayısı kıpırdamıyor. İkinci yapıştırma etkisizleştirilmiş demektir.

Üçüncü kontrol worker'ı kasten öldürmektir. İş yarısına geldiğinde süreci durdurun, tekrar başlatın ve processed alanının geriye gitmediğini, tablodaki satır sayısının da artmadığını görün. Bu testi bir kez elle yapmak, devam edebilirliğin gerçekten çalıştığına dair kâğıt üstündeki her iddiadan daha ikna edicidir ve yapması bir dakika sürer.

Sonra işin bitmesini izleyin ve sonucun tam olduğunu doğrulayın:

SELECT j.total, j.processed, j.state,
       (SELECT count(*) FROM contacts WHERE list_id = j.list_id)          AS in_table,
       (SELECT count(DISTINCT msisdn) FROM contacts WHERE list_id = j.list_id) AS distinct_rows
FROM jobs j WHERE j.id = $1;

processed ile total eşit, in_table ile distinct_rows eşit olmalı. Yapıştırılan listede tekrar varsa in_table değeri total değerinden tam o kadar düşük çıkar; iş raporu bunu arayüzde açıkça yazmalı, kullanıcıyı eksik satırların nereye gittiğini merak eder hâlde bırakmamalı.

Nelere dikkat etmeli

• Proxy'deki istek boyutu sınırı, timeout'tan ayrı bir ayardır. Üç megabaytlık bir yapıştırma, sınır yükseltilmediyse handler'a hiç ulaşmadan reddedilir ve çıkan hata timeout'a hiç benzemez.
• Ayrıştırılmış sonucu değil ham veriyi saklayın. En çok değişecek parça ayrıştırmadır; saklanan ham veri, parser hatasını düzelttikten sonra işi yeniden işlemenizi sağlar.
• Reddedilen satırları satır numarası ve gerekçesiyle, kullanıcının indirebileceği bir dosya olarak raporlayın. Yüz binin doksan sekiz binini sessizce alan bir iş, yüksek sesle hata veren işten kötüdür.
• İşler tablosuna daha ilk günden saklama süresi koyun. Ham veriler büyüktür ve üç megabaytlık blob'ların budanmadığı bir tablo, şemadaki her şeyden hızlı biçimde disk sorununa dönüşür.
• Süreç yeniden başladığında çalışır durumdaki bir iş tekrar ele alınmalı, yani worker da servisin geri kalanı gibi yeniden başlatmadan sağ çıkmalı.

Buradaki genel biçim saklamaya değer. Boyutunu kullanıcının seçtiği her işlem, er ya da geç önündeki süre sınırını aşacaktır ve istek döngüsü bunu öğrenmek için yanlış yerdir. Girdiyi kabul edin, kalıcı olarak yazın, kullanıcının izleyebileceği bir şeyle yanıt verin ve işi, hatanın yarım yazılmış bir tabloya değil yalnızca bir tekrara mal olduğu yerde yapın. Yavaş bir gönderim hattını hızlıya çeviren mantık da aynı: küçük birimler, kayıtlı bir konum ve her an durdurulabilen bir worker. Kapasitenin aslında nereden geldiğinin hikâyesinin büyük kısmı budur.