Arka plan işleri: kuyruk, zamanlayıcı ve olaylar

Her ajan çalıştırması bir HTTP isteğiyle başlamaz. Birinin biriken işi işlemesi, 03:00'te tetiklenmesi ya da bir olaya tepki vermesi gerekir — ve bunların her biri, bir çökmenin yan etkiyi ikileyebileceği bir yerdir. Bu üç paket, aynı garantiyi bu üç sınıra koyar.

Hiçbiri dayanıklılığı yeniden yazmaz. Bir iş işleyicisi tipik olarak runDurable çağırır, bir tetikleyici dayanıklı bir workflow koşar, bir olay tüketicisinden de aynısı beklenir. Her paketin eklediği şey, journal'ın tek başına veremeyeceği parçadır: iki worker'ın aynı işi almaması için bir kilit, bir cron aralığının iki kez koşmaması için bir tetikleme kaydı, ve bir olayın bitmeden bitti işaretlenmemesi için bir ack işareti.

enqueue işi yazar, createWorker onu kapıp koşturur. Bir worker TTL'li bir kilit tutar; çökerse kilit bayatlar, başka bir worker işi geri alır, işleyici tekrar koşar — ve işleyici aynı runId'yi devam ettirdiği için yan etki yine bir kez olur. Başarısızlık maxAttempts ile sınırlıdır; sonrasında iş dead-letter olur ve retryJob ile diriltilebilir.

Boş-yoklama geri çekilmesi varsayılan olarak açıktır: bir yoklama hiçbir şey kapamazsa aralık bir tavana kadar ikiye katlanır ve iş kapılır kapılmaz sıfırlanır. Bu varsayılanın sebebi bir denetim bulgusudur — boş bir kuyruğu sabit aralıkla yoklayan bin tüketici saniyede on binlerce sorgu üretir.

İşler kuruluşlar arasında adil alınır. Her kuruluşun işleri bir şerittir; worker, depoya en uzun süredir bekleyen kuruluşun sıradaki işini sorar ve tek bir depo çağrısında kilitler (Postgres ve MySQL'de SKIP LOCKED, MongoDB'de findOneAndUpdate, SQLite ve libSQL'de tek yazıcı). Boş bir poll, kuruluş sayısı ne olursa olsun sabit sayıda sorgudur. Hâlâ 0.16 kuyruğu tutan bir depo, bu sürümün worker'ları başlamadan önce gnl upgrade ister.

import { enqueue, createWorker } from '@gnldev/queue';

await enqueue(storage.work, 'reindex', { docId: 'd-9' });

// one handler per job type; ctx carries the job's own durable runId and journal
const worker = createWorker(
  storage,
  {
    reindex: async (payload, ctx) => {
      // a reclaim resumes this runId, it does not restart it
      await runDurable({ runId: ctx.runId, journal: ctx.journal, model, tools, prompt: payload.docId });
    },
  },
  { ttlMs: 30_000, maxAttempts: 3 },
);

worker.start();

enqueueMany(work, jobs) çok sayıda işi tek yazımda ekler — ya hepsi ya hiçbiri. Her iş type, payload ve isteğe bağlı options taşıyan bir nesnedir; options, enqueue'un aldıklarıdır ve iş enqueue'un denetlediği gibi denetlenir: geçersiz tek bir iş, hiçbir şey yazılmadan çağrıyı reddettirir. Kimlikler çağrı sırasıyla döner; daha önce eklenmiş ya da çağrıda tekrarlanan bir kimlik tek iştir; maxDepth çağrının kendi işlerini sayar. Adalet, enqueue'un işleri birbiri ardına eklemesininkiyle aynıdır, ama çağrı her kuruluşun sırasını işin başına değil bir kez alır — büyük bir toplu işi tek bir kuruluşa koymanın yolu, örneğin bir CSV içe aktarımı (Postgres'te 30.000 iş yaklaşık 3 saniyede).

listJobs bir sayfa döndürür: .items'ı okuyun, varsa nextCursor'ı da. orgId verilmezse yalnızca kuruluşu olmayan işleri listeler, her kuruluşunkini değil. Depodaki her işi görmek isteyen operatör listAllJobs kullanır: önce kuruluşu olmayan işler, sonra her kuruluşun şeridi ad sırasıyla, her şerit ekleme sırasıyla. Bir listeleme başladığı anda sabitlenir — ilk sayfasından önce eklenen her iş tam bir kez listelenir, sonra eklenen sonraki listelemeye aittir — bu yüzden hiç durmayan üreticiler altında bile biter.

import { enqueueMany, listJobs, listAllJobs } from '@gnldev/queue';

// a CSV import: one write, all or none; the ids come back in call order
const ids = await enqueueMany(
  storage.work,
  rows.map((row) => ({
    type: 'import-row',
    payload: row,
    options: { id: `import:${row.id}`, orgId: 'acme' },
  })),
);

// one organization's jobs, a page at a time
const page = await listJobs(storage.work, { orgId: 'acme', limit: 50 });
page.items;       // the jobs
page.nextCursor;  // pass it back as { cursor } for the next page

// an operator's view: every job on the store
const all = await listAllJobs(storage.work, { limit: 100 });

scheduleWorkflow bir tetikleyici kaydeder: tek seferlik at, periyodik every ya da 5 alanlı cron (UTC, dakika çözünürlüğü). Tetikleyici tanımı journal'da değişmezdir, yalnız durumu hareket eder. Her tetikleme bir çalıştırma kilidi alır ve runId'sini tetikleme sayacından türetir; böylece aynı aralık iki kez başlayamaz.

Zaman veri olarak ele alınır: sonraki çalışma zamanı çözülür ve sonra journal'a dondurulur — bir replay'in duvar saatiyle kaymasını engelleyen şey budur.

misfire, hiçbir poller çalışmadığında kaçırılmış bir pencerenin ne anlama geldiğine karar verir. 'latest' (varsayılan) kaçırılan son oluşumu bir kez tetikler; 'earliest' en eskisini bir kez; 'none' kaçırılan hiçbir oluşumu çalıştırmaz; 'catchup' kaçırılan her oluşumu, en eskiden başlayarak, poll başına bir tane tetikler. Her değerde takvim, şimdiden sonraki ilk oluşumdan planlanan ızgarasında devam eder, böylece kayma birikmez. Gece raporu genelde latest, fatura döngüsü genelde catchup ister.

'catchup' varsayılan olarak sınırsızdır — dakikalık bir cron'da bir gün uyumak 1.440 tetikleme demektir. maxCatchup: n ile yalnızca kaçırılan en son n oluşum tetiklenir ve tetikleyici, kaç tanesinin atlandığını ve en eskisinin zamanını söyleyen tek bir uyarı yazar. Her tetikleme hangi oluşum için olduğunu bilir: bir workflow adımında ctx.scheduledFor (takvimdeki yuva, poll'un çalıştığı an değil); yeniden denemelerde ve devam ettirmelerde korunur.

Zamanı gelen tetikleyiciler artık birbirini beklemez. Zamanı gelen her tetikleyici hemen tetiklenir ve concurrency'nin (createScheduler ve pollScheduler üzerinde) varsayılan bir sınırı yoktur — sınırlamak için concurrency: n verin; 10 saniyelik bir workflow eskiden tüm poll döngüsünü tutar ve 2 saniyelik bir cron'u 12 tetiklemeden 2'ye düşürürdü. Bir tetiklemenin kilidi alınır ve kararı onun altında depo saatiyle denetlenir; saati kaymış bir süreç artık aynı oluşumu iki kez başlatmaz.

import { scheduleWorkflow, createScheduler } from '@gnldev/scheduler';

await scheduleWorkflow(journal, {
  name: 'nightly-report',
  cron: '0 3 * * *',       // 03:00 UTC
  misfire: 'latest',       // the default: after an outage, the last missed slot fires once
  // misfire: 'catchup', maxCatchup: 10,   // or every missed slot, capped at the 10 latest
});

createScheduler(journal, runner, { concurrency: 4 }).start(); // concurrency: no limit unless you cap it

// inside a workflow step: the occurrence this fire is for
//   ctx.scheduledFor   // epoch ms of the slot, not of the poll

emit bir log'a ekler; her tüketici olayları kendi ack işareti altında işler, yani N tüketicinin her biri her olayı görür. İşaret, işleyici başarılı olduktan sonra bir CAS ile yazılır. Bütün tasarım bu sıralamadadır: işleyici hata fırlatırsa ya da süreç ölürse olay kaybolmaz, sonraki yoklama tekrar dener.

Bedeli gizlenmez, yazılır: bu exactly-once işaretleme, at-least-once teslimtir. Başarılı bir işleyici ile ack'i arasındaki bir çökme — ya da eşzamanlı bir yoklama yarışı — tekrar teslime yol açabilir. İşleyiciyi idempotent yazın ya da içine runDurable / claim koyun; o zaman tekrar teslimin bir maliyeti kalmaz.

import { emit, createConsumer } from '@gnldev/events';

await emit(storage.work, 'order.paid', { orderId: 'o-7' });

const consumer = createConsumer(
  storage.work,
  'order.paid',
  async (payload, meta) => {
    // the ack marker is written only after this returns — so make it idempotent
    await runDurable({ runId: `evt:${meta.id}`, journal, model, tools, prompt: payload.orderId });
  },
  { name: 'fulfilment' },   // ack markers are separated by this name (fan-out)
);

consumer.start();

Bir handler'ın çalışma hakkı, o çalışırken bitebilir: iş iptal edilir, kiralama kaybedildiği için artık başka bir worker onu tutar ya da handler işi izin verilen kadar uzun tutmuştur. Üçü de handler'a tek bir şey olarak ulaşır: ctx.signal, bir ExecutionRevokedError ile abort olur; why hangisi olduğunu söyler: canceled, lease-lost ya da timeout. Eskiden yalnızca iptalde abort olurdu ve kiralamasını kaybeden bir worker yeni sahibinin yanında çalışmayı sürdürürdü.

Kuyruk worker'ındaki maxHoldMs (varsayılan: sınır yok) süre sınırıdır. Dolduğunda ctx.signal timeout ile abort olur ve deneme başarısız sayılır — yeniden denenir, maxAttempts'te dead-letter olur. Kiralama handler dönene kadar tutulur, yani yanında hiçbir şey çalışmaz. Bir olay handler'ı aynısını meta.signal ile alır (timeout, lease-lost); createConsumer de maxHoldMs alır. Bir zamanlayıcı tetiklemesinin runner'ı opts.signal alır; kilidin kendisine ait olmadığını bir yenileme bildirdiğinde ya da art arda üç yenileme cevapsız kaldığında lease-lost olarak abort olur; createGnl().runWorkflow buna bakarak workflow'u bir sonraki adım sınırında durdurur — durdurur, iptal etmez.

createWorker(
  storage,
  {
    reindex: async (payload, ctx) => {
      // ctx.signal.reason is an ExecutionRevokedError; its why is 'canceled' | 'lease-lost' | 'timeout'.
      // ctx.run passes the signal on to the run; other long work checks it itself.
      const res = await fetch(payload.url, { signal: ctx.signal });
      return ctx.run({ model, tools, prompt: await res.text() });
    },
  },
  { ttlMs: 30_000, maxAttempts: 3, maxHoldMs: 10 * 60_000 }, // a handler held past 10 minutes: timeout
);

Framework, sinyali yok sayan kodu durduramaz. ctx.run sinyali run'a iletir; diğer uzun işler onu kendisi kontrol etmelidir ve yan etkiler ctx.run'ın içinde durmalıdır ki devralma onları tekrarlamasın. Kiralaması kaybedilmiş ya da maxHoldMs'i dolmuş bir handler hâlâ çalışıyorsa, ttlMs (kuyruk), leaseMs (olaylar) ya da lockTtlMs (zamanlayıcı) sonra bir kez uyarılır (console.warn). Bir olayın meta'sını toEqual ile karşılaştıran bir test artık toMatchObject ister.

Bir tüketicinin imleci, ancak işleyicisinin kabul ettiği olayın ötesine geçer. Sürekli hata fırlatan bir işleyici bu yüzden imleci sonsuza kadar park ediyordu: hatalı olayın arkasındaki her olay, asla başarılı olmayacak bir teslimatı bekliyordu. Kaybolmuş mesaj değil — durmuş bir konu.

Artık bir teslimat maxAttempts başarısızlıktan sonra (varsayılan 8) karantinaya alınıyor ve imleç ilerliyor. Denemeler retryDelayMs ile aralanıyor — 60 saniyeden başlayıp ikiye katlanan, bir saatte sınırlanan bir üstel dizi — böylece bütçe yoklama aralığında bir saniyede tükenmek yerine yaklaşık iki saate yayılıyor. Bir yeniden deneme bütçesi, ancak var olduğu kesintiden uzun yaşarsa bütçedir.

Vakti gelmemiş bir olay atlanır ama imleci yine dondurur: hâlâ yeniden denenebilir olduğu için arkasındakiler geçilmiş sayılamaz. Beklemek, vazgeçmek değildir. Hiç karantinaya almak istemiyorsanız maxAttempts: Infinity bu şekli kalıcı yapar — diğer olaylar yine akar, ama bu tüketici park kalır ve her yoklamada oradan itibaren yeniden tarar.

import { createConsumer, listDeadEvents, retryDeadEvent } from '@gnldev/events';

const consumer = createConsumer(storage.work, 'order.paid', handler, {
  name: 'fulfilment',
  maxAttempts: 8,        // quarantine after this many FAILED attempts (the default)
  retryDelayMs: 60_000,  // fixed spacing; omit for exponential 60s → 1h (the default)
});

// what is parked, and why
const dead = await listDeadEvents(storage.work, 'order.paid', 'fulfilment');
// [{ id: 'e-3', attempts: 8, error: 'ECONNREFUSED', at: 173... }]

// hand one back once the downstream is healthy again
await retryDeadEvent(storage.work, 'order.paid', 'fulfilment', 'e-3');

listDeadEvents(work, topic, consumer) parkta ne olduğunu gösterir, retryDeadEvent(work, topic, consumer, eventId) birini geri verir. Studio ikisini de sunar, operatörün betik yazmasına gerek yok. Adresleme üçlü ile yapılır: (topic, consumer, id) — bir konu dallanır, yani tek bir olayın tüketici başına bir karantina kaydı olur ve tek başına id bunların birkaçını birden gösterir.