Aszinkron munka: Queues, Cron, Workflows, Durable Objects

7.1Queues: az SQS-ed megfelelője — polling nélkül

A Queues fogalmilag ismerős: üzenetek, batchelés, retry, dead letter queue. A strukturális különbség az SQS-hez képest, hogy nincs consumer-infrastruktúra és nincs polling: a consumer egy queue handler egy Workerben (2. modul), amit a platform hívogat a batchekkel, és a párhuzamos fogyasztók számát automatikusan skálázza a torlódáshoz.

A konfiguráció — és hogy mit jelentenek a számok

{
  "queues": {
    "producers": [{ "binding": "JOBS", "queue": "jobs" }],
    "consumers": [{
      "queue": "jobs",
      "max_batch_size": 50,        // legfeljebb ennyi üzenet egy batchben…
      "max_batch_timeout": 5,      // …vagy ennyi mp után indul, ami hamarabb teljesül
      "max_retries": 5,
      "retry_delay": 120,          // alapértelmezett várakozás újrapróbálás előtt (mp)
      "max_concurrency": 10,       // felső korlát a párhuzamos consumereknek (opcionális)
      "dead_letter_queue": "jobs-dlq"
    }]
  }
}

A max_batch_size / max_batch_timeout páros a klasszikus áteresztés-vs-latency trade-off: nagy batch + hosszabb timeout = kevesebb Worker-hívás és jobb költség; kicsi batch + rövid timeout = gyorsabb reakció. Emailküldésnél az 50/5s remek; „a user rákattintott és várja az eredményt" jellegű munkánál 10/1s körül járj.

1. példa: számla-email — a legegyszerűbb minta

A cél: a checkout API-d ne várjon az emailküldő külső API-jára. A producer-oldal egyetlen sor, a válasz azonnal megy a usernek:

export default defineEventHandler(async (event) => {
  const { env } = event.context.cloudflare;
  const order = await createOrder(env, await readBody(event));

  // nem await-oljuk az emailküldést — a sor majd gondoskodik róla
  await env.JOBS.send({ type: "invoice-email", orderId: order.id });

  return { ok: true, orderId: order.id };   // azonnali válasz
});

2. példa: consumer részleges hibával — üzenetenkénti ack

Ez a Queues egyik legfontosabb mintája. Egy 50-es batchből mondjuk 48 sikerül, 2 nem — nem akarod az egész batchet újrafuttatni (az SQS-nél ez volt a visibility-timeout fejfájás). Üzenetenként döntesz:

export default {
  async queue(batch, env, ctx) {
    for (const msg of batch.messages) {
      try {
        switch (msg.body.type) {
          case "invoice-email": await sendInvoice(env, msg.body.orderId); break;
          case "thumbnail":     await makeThumb(env, msg.body.key);       break;
        }
        msg.ack();                              // ✓ kész, soha többé nem jön
      } catch (err) {
        if (isRateLimit(err)) {
          msg.retry({ delaySeconds: 600 });     // külső API fullad → 10 perc múlva
        } else {
          msg.retry();                          // default retry_delay-jel
        }
      }
    }
    // amire se ack, se retry nem hívódott, azt a platform retry-olja
  },
};

3. példa: a poison message és a DLQ-consumer

Egy hibás üzenet (rossz formátum, törölt rekordra hivatkozik) örökké elbukna — a max_retries után a platform átteszi a jobs-dlq sorba. A DLQ ne temető legyen, hanem postafiók: tegyél rá egy saját consumert, ami riaszt és eltárolja a hullákat vizsgálatra:

async queue(batch, env) {
  for (const msg of batch.messages) {
    await env.DB.prepare(                       // D1-be naplózzuk vizsgálatra
      "INSERT INTO failed_jobs (body, failed_at) VALUES (?, ?)"
    ).bind(JSON.stringify(msg.body), new Date().toISOString()).run();
    msg.ack();                                  // a DLQ-ból már nem retry-olunk
  }
  await notifySlack(env, `${batch.messages.length} üzenet halt be a jobs sorban`);
}

Amit az SQS-hez képest tudnod kell

7.2Cron Triggers: az ütemezett munkák

A scheduled handler (2. modul) + egy sor konfig — az EventBridge Scheduler + külön Lambda megfelelője. A cron kifejezések UTC-ben értendők (a 3:00 budapesti hajnali takarítás nyáron 0 1 * * *!), és több cron is routolható egy Workeren belül:

{ "triggers": { "crons": ["0 1 * * *", "*/15 * * * *"] } }

async scheduled(event, env, ctx) {
  switch (event.cron) {
    case "0 1 * * *":    return nightlyJobs(env);
    case "*/15 * * * *": return syncExchangeRates(env);
  }
}

A fontos minta: a cron indító, nem munkás

A cron-futásra is CPU-limit vonatkozik, és ha félúton elhal, nincs beépített folytatás. Ezért az ökölszabály: a scheduled handler csak szétosztja a munkát, a tényleges melót a Queues vagy a Workflows végzi — így kapsz retry-t, láthatóságot és skálázást. Így néz ki a „napi digest email minden tenantnak" a te SaaS-odban:

async scheduled(event, env, ctx) {
  // 1. cron: kiolvassuk a címzetteket a registryből (5b. modul)
  const tenants = await env.REGISTRY
    .prepare("SELECT id FROM tenants WHERE status = 'active'").all();

  // 2. tenantonként egy üzenet — max 100-asával batchelt küldés
  const msgs = tenants.results.map(t =>
    ({ body: { type: "daily-digest", tenantId: t.id } }));
  for (let i = 0; i < msgs.length; i += 100)
    await env.JOBS.sendBatch(msgs.slice(i, i + 100));

  // 3. a consumer tenantonként, retry-jal, párhuzamosan dolgozik
}

Lokálisan tesztelni: wrangler dev --test-scheduled, majd curl "http://localhost:8787/__scheduled?cron=0+1+*+*+*" — nem kell hajnali egyig várnod.

7.3Workflows: a Step Functions, ha kódban írhatnád

A Step Functions-höz képest a leglényegesebb különbség: nincs ASL JSON, nincs vizuális state machine-szerkesztő — a folyamat egy sima TypeScript-osztály. A platform durable execution-t garantál: minden step.do eredménye perzisztálódik, és bármilyen hiba vagy újraindulás után az instance a legutóbbi sikeres lépés utáni ponttól folytatódik.

A mentális modell: a run() újrajátszódik, a lépések nem

Ezt az egy dolgot kell igazán megérteni, minden más ebből következik. Amikor egy instance folytatódik (retry, alvás utáni ébredés, platform-újraindulás), a run() metódus az elejétől újra lefut — de a már teljesített step.do-k nem futnak le újra, hanem a perzisztált eredményüket adják vissza azonnal. Ebből következik a három vasszabály:

• Minden mellékhatás step-be való. Ami a run()-ban a stepeken kívül van, az minden újrajátszáskor lefut — egy stepen kívüli sendEmail() ötször küldheti el ugyanazt a levelet.
• A step eredménye szerializálható legyen (JSON-ba írható) — ez kerül a tartós tárba, és ezt kapod vissza replay-kor.
• A step legyen a retry egysége. Egy „nagy step sok mindennel" hibánál az egészet ismétli; sok kis step finomabb újrapróbálást és jobb láthatóságot ad. Ökölszabály: egy külső hívás ≈ egy step.

1. példa: tenant-onboarding — az 5b. modul köre bezárul

import { WorkflowEntrypoint } from "cloudflare:workers";
import { NonRetryableError } from "cloudflare:workflows";

export class OnboardTenant extends WorkflowEntrypoint<Env, { slug: string, region: string }> {
  async run(event, step) {
    const { slug, region } = event.payload;

    // 1) D1 létrehozása a tenant régiójában (5b.2!) — REST API-val
    const dbId = await step.do("create tenant db", async () => {
      const db = await createD1(this.env, `tenant-${slug}`, region);
      return db.uuid;                       // szerializálható → perzisztálódik
    });

    // 2) séma felhúzása — ha elhal, CSAK ez ismétlődik, a create nem
    await step.do("apply schema",
      { retries: { limit: 5, delay: "10 seconds", backoff: "exponential" } },
      () => queryTenant(this.env, dbId, SCHEMA_SQL, []));

    // 3) registry-bejegyzés — üzletileg érvénytelen slug? ne is próbálkozzunk újra
    await step.do("register", async () => {
      const ok = await registerTenant(this.env, slug, dbId, region);
      if (!ok) throw new NonRetryableError("slug already taken");
    });

    // 4) welcome email — mehet sorba is, de itt látszik a folyamatban
    await step.do("welcome email", () => sendWelcome(this.env, slug));
  }
}

Figyeld meg a három retry-viselkedést: a create tenant db a default szabályokkal megy; az apply schema saját, exponenciális backoffot kap; a register pedig NonRetryableError-ral azonnal, végleg elbukik, ha az ok üzleti (nem átmeneti) — ez a „retry storm" ellenszere.

2. példa: várakozás külső eseményre — a webhook-minta

async run(event, step) {
  await step.do("start trial", () => activateTrial(this.env, event.payload.tenantId));

  try {
    // max 14 napig várunk arra, hogy előfizet — a webhook-endpoint ébreszt:
    //   const wf = await env.ONBOARD.get(instanceId);
    //   await wf.sendEvent({ type: "subscribed", payload: { plan } });
    const sub = await step.waitForEvent("wait for subscribe",
      { type: "subscribed", timeout: "14 days" });
    await step.do("activate paid plan", () => activate(this.env, sub.payload.plan));
  } catch {
    // timeout: nem fizetett elő — lezárjuk szépen
    await step.do("downgrade", () => downgradeToFree(this.env, event.payload.tenantId));
    await step.do("feedback email", () => sendFeedbackAsk(this.env, event.payload.tenantId));
  }
}

Vedd észre, mi nincs ebben a kódban: nincs „trial_expires_at" oszlop + napi cron, ami lejáratokat vadászik + állapotgép a DB-ben. A 14 napos üzleti folyamat egyetlen, olvasható függvény — és az alvás alatt nem fut és nem fizetsz semmit. Kiegészítésül van step.sleepUntil(dátum) fix időpontra, és saga-jellegű rollback is: a step.do-hoz kompenzáló handlert regisztrálhatsz (készlet-feloldás, refund), ami későbbi hibánál fordított sorrendben fut le.

Indítás, követés, üzemeltetés

// indítás a Workerből / API route-ból
const instance = await env.ONBOARD.create({
  id: `onboard-${slug}`,               // determinisztikus ID = dedup ingyen!
  params: { slug, region: "weur" },
});

// státusz bármikor: fut? melyik stepnél? mi volt a hiba?
const status = await (await env.ONBOARD.get(`onboard-${slug}`)).status();

A determinisztikus instance-ID egy rejtett gyöngyszem: ha a „subscribe" webhook kétszer érkezik és kétszer próbálnád indítani ugyanazt az onboardingot, a második create ugyanarra az ID-re elutasítódik — beépített dedup. A dashboardon (és wranglerrel) instance-onként látod a lépések állapotát, időzítését, hibáit — ez a Step Functions execution-history megfelelője.

7.4Durable Objects: az eszköz, aminek nincs AWS-megfelelője

Az 5b. modulból már tudod, mi ez a motorháztető alatt — most nézzük fejlesztői szemmel. A DO három tulajdonság együttese, és pont ez a kombináció nincs meg az AWS-en:

• Címezhetőség: névből cím — idFromName("acme") a világon mindenhonnan ugyanahhoz a példányhoz vezet. Nem kell service discovery, nem kell routing-tábla.
• Egyszálúság: a példányon belül nincs párhuzamosság — a „check-then-act" logikád (van-e még hely? akkor foglald!) verseny nélkül, lockok nélkül helyes.
• Kolokált állapot: memória + saját SQLite-tár (10 GB) közvetlenül az objektumban — a „lekérdezés" gyakran memória-hozzáférés, nem hálózati kör.

1. példa: rate limiter tenantonként

import { DurableObject } from "cloudflare:workers";

export class RateLimiter extends DurableObject {
  async check(limit = 100) {
    const w = currentMinute();
    const row = this.ctx.storage.sql
      .exec("SELECT count FROM hits WHERE window = ?", w).one();
    if ((row?.count ?? 0) >= limit) return false;
    this.ctx.storage.sql.exec(
      "INSERT INTO hits(window,count) VALUES(?,1) ON CONFLICT(window) DO UPDATE SET count=count+1", w);
    return true;
  }
}

// a Workerben — névre címezve, RPC-hívásként:
const limiter = env.RATE_LIMITER.get(env.RATE_LIMITER.idFromName(tenantId));
if (!(await limiter.check(plan.rateLimit)))
  return new Response("Too many requests", { status: 429 });

Miért nem jó erre semmi más a palettáról? A KV eventually consistent és ~1 írás/s/kulcs — a számláló szétcsúszna. A D1-nél minden check egy hálózati kör a primary-ba — lassú és terheli a DB-t. A DO-nál a tenant limitere egy helyen, memóriaközel számol.

2. példa: készletfoglalás — a D1 „nincs interaktív tranzakció" rés betömése

Az 5. modulban függőben maradt a kérdés: mi van, ha a „read → üzleti döntés kódban → write" mintához tényleg erős konzisztencia kell? Íme a válasz — a jegyfoglalás tankönyvi példáján:

export class EventSeats extends DurableObject {
  async reserve(userId: string, qty: number) {
    // egyszálú vagyok: e két sor közé SENKI nem ékelődhet be
    const free = this.ctx.storage.sql
      .exec("SELECT free FROM seats").one().free;
    if (free < qty) return { ok: false, free };

    this.ctx.storage.sql.exec("UPDATE seats SET free = free - ?", qty);
    this.ctx.storage.sql.exec(
      "INSERT INTO holds(user_id, qty, expires_at) VALUES (?,?,?)",
      userId, qty, Date.now() + 600_000);

    // alarm: 10 perc múlva ébredek és felszabadítom a lejárt holdokat
    await this.ctx.storage.setAlarm(Date.now() + 600_000);
    return { ok: true };
  }

  async alarm() {   // a DO saját magának időzít — nincs külön cron!
    this.ctx.storage.sql.exec(
      "UPDATE seats SET free = free + (SELECT COALESCE(SUM(qty),0) FROM holds WHERE expires_at < ?)", Date.now());
    this.ctx.storage.sql.exec("DELETE FROM holds WHERE expires_at < ?", Date.now());
  }
}

Két dolgot mutat ez a példa. Egy: az egyszálúság miatt a SELECT és az UPDATE közötti üzleti logika helyesen versenymentes — az, amit Postgresben SELECT … FOR UPDATE-tel vagy serializable tranzakcióval csináltál. Kettő: az alarm — a DO saját magának időzíthet jövőbeli futást, így az „elengedett kosár / lejárt hold takarítás" nem igényel külön cron-t és sweep-query-ket: minden objektum a saját szemetét takarítja.

3. példa: élő jelenlét (presence) WebSockettel

export class DocPresence extends DurableObject {
  async fetch(request) {
    const [client, server] = Object.values(new WebSocketPair());
    this.ctx.acceptWebSocket(server);          // hibernation: alvó kapcsolat ≈ ingyen
    return new Response(null, { status: 101, webSocket: client });
  }

  async webSocketMessage(ws, msg) {
    // broadcast mindenkinek, aki ugyanezt a dokumentumot nézi
    for (const peer of this.ctx.getWebSockets())
      if (peer !== ws) peer.send(msg);
  }
}

// routing a Workerből: dokumentumonként EGY objektum
const doc = env.PRESENCE.get(env.PRESENCE.idFromName(docId));
return doc.fetch(request);   // WebSocket upgrade továbbadása

AWS-en ugyanez: API Gateway WebSocket + Lambda + DynamoDB a connection-táblának + fan-out logika. Itt a „szoba" maga az objektum: a kapcsolatlista az állapota, a broadcast egy for-ciklus. A hibernation a költség-kulcs: a csendben ülő kapcsolatok alatt az objektum kikerül a memóriából, és nem fizetsz futásidőt — tízezres kapcsolatszám is olcsó.

7.5Melyiket mikor? — szcenáriókon keresztül

7.6Ellenőrizd magad