Build an idempotent scheduled job with Cloudflare Queues and D1

A cron trigger can start a job on schedule. It cannot guarantee the job runs only once.

Cloudflare Queues can deliver the same message more than once. A Worker can also stop after completing the work but before acknowledging the message.

This is normal for an at-least-once system.

In this tutorial we’ll build a scheduled job that checks pricing pages. Each page has its own interval, such as every 24 hours.

We’ll use:

• a cron trigger to find due checks
• Cloudflare Queues to run them in the background
• D1 to prevent duplicate work
• an expiring lease to recover from crashed Workers
• a fixed due slot to preserve the original schedule

The same pattern works for reports, imports, cleanup jobs, and API synchronization.

Why a cron trigger is not enough

Suppose we check a pricing page every day at 08:00.

The cron runs and adds a message to the queue. The queue consumer fetches the page and saves the result.

Then the Worker stops before calling message.ack().

Cloudflare does not know the work finished. It delivers the message again.

If our code just runs the job again, we could save two checks or send two alerts.

We need to make the work idempotent. Processing the same scheduled check twice must have the same durable result as processing it once.

Give every scheduled check an identity

Do not use the time when the consumer starts as the job identity.

Use the time when the work was due. I call this the due slot.

For example, this target and due slot always produce the same work ID:

function workId(targetId: string, scheduledFor: number) {
  return `${targetId}:${new Date(scheduledFor).toISOString()}`
}

const id = workId('hetzner-cloud-pricing', 1785916800000)

Every duplicate delivery now carries the same identity.

The queue message includes both values because they have different jobs:

type CheckMessage = {
  kind: 'check-source'
  targetId: string
  workId: string
  scheduledFor: number
}

targetId tells us what to check. scheduledFor anchors the schedule. workId identifies this exact piece of work.

Store the schedule in D1

The cron trigger should not own the schedule. D1 should.

Here is the small schema we’ll use:

CREATE TABLE monitor_targets (
  id TEXT PRIMARY KEY NOT NULL,
  url TEXT NOT NULL,
  interval_minutes INTEGER NOT NULL,
  enabled INTEGER NOT NULL DEFAULT 1,
  next_check_at INTEGER NOT NULL,
  lease_token TEXT,
  lease_expires_at INTEGER,
  CHECK (interval_minutes >= 15),
  CHECK (
    (lease_token IS NULL AND lease_expires_at IS NULL)
    OR
    (lease_token IS NOT NULL AND lease_expires_at IS NOT NULL)
  )
);

CREATE INDEX monitor_targets_schedule_idx
ON monitor_targets (enabled, next_check_at);

Dates are Unix timestamps in milliseconds.

Each target stores its next due slot. It also has a lease token and lease expiration time. We’ll use those fields in the consumer.

Now create a table for completed and failed checks:

CREATE TABLE source_checks (
  work_id TEXT PRIMARY KEY NOT NULL,
  target_id TEXT NOT NULL,
  scheduled_for INTEGER NOT NULL,
  checked_at INTEGER NOT NULL,
  status TEXT NOT NULL,
  attempt_count INTEGER NOT NULL DEFAULT 1,
  result_hash TEXT,
  error_message TEXT,
  FOREIGN KEY (target_id) REFERENCES monitor_targets(id),
  CHECK (status IN ('unchanged', 'changed', 'failed')),
  CHECK (attempt_count > 0)
);

CREATE UNIQUE INDEX source_checks_target_schedule_unique
ON source_checks (target_id, scheduled_for);

The primary key protects the deterministic work ID.

The unique index adds a second guard. Even if a bug creates a different work ID, D1 still allows only one row for a target and due slot.

Configure the cron and queue

Let’s configure a Cloudflare cron trigger that runs every 15 minutes:

{
  "triggers": {
    "crons": ["*/15 * * * *"]
  },
  "queues": {
    "producers": [
      {
        "binding": "MONITOR_QUEUE",
        "queue": "pricing-monitor"
      }
    ],
    "consumers": [
      {
        "queue": "pricing-monitor",
        "max_batch_size": 10,
        "max_batch_timeout": 5,
        "max_retries": 5,
        "dead_letter_queue": "pricing-monitor-dlq"
      }
    ]
  },
  "d1_databases": [
    {
      "binding": "DB",
      "database_name": "pricing-monitor",
      "database_id": "00000000-0000-0000-0000-000000000000"
    }
  ]
}

The cron frequency and target frequency are separate.

The cron looks for work every 15 minutes. One target can run hourly while another runs daily.

Enqueue the due targets

The scheduled handler queries D1 for targets whose due time has passed:

type DueTarget = {
  id: string
  next_check_at: number
}

async function enqueueDueChecks(env: Env, now: number) {
  const result = await env.DB.prepare(`
    SELECT id, next_check_at
    FROM monitor_targets
    WHERE enabled = 1 AND next_check_at <= ?
    ORDER BY next_check_at
    LIMIT 100
  `)
    .bind(now)
    .all<DueTarget>()

  if (result.results.length === 0) {
    return
  }

  await env.MONITOR_QUEUE.sendBatch(
    result.results.map(target => ({
      body: {
        kind: 'check-source',
        targetId: target.id,
        workId: workId(target.id, target.next_check_at),
        scheduledFor: target.next_check_at,
      } satisfies CheckMessage,
    }))
  )
}

Then call it from the Worker scheduled handler:

export default {
  async scheduled(_controller, env, _context) {
    await enqueueDueChecks(env, Date.now())
  },
} satisfies ExportedHandler<Env, CheckMessage>

Notice that enqueuing does not advance next_check_at.

The cron might find the same target again before a consumer finishes. That’s fine. The consumer will use D1 to decide which message can proceed.

Claim the due slot with a lease

We don’t want two consumers fetching the same page at once.

Before doing any slow work, each consumer tries to claim the target:

async function claimTarget(
  env: Env,
  message: CheckMessage,
  now: number
) {
  const token = crypto.randomUUID()
  const leaseExpiresAt = now + 5 * 60_000

  const claimed = await env.DB.prepare(`
    UPDATE monitor_targets
    SET lease_token = ?, lease_expires_at = ?
    WHERE id = ?
      AND enabled = 1
      AND next_check_at = ?
      AND (
        lease_token IS NULL
        OR lease_expires_at <= ?
      )
    RETURNING id
  `)
    .bind(
      token,
      leaseExpiresAt,
      message.targetId,
      message.scheduledFor,
      now
    )
    .first<{ id: string }>()

  if (!claimed) {
    return null
  }

  return { token, leaseExpiresAt }
}

The important part is the conditional UPDATE.

D1 changes the row only when all conditions still match. Two consumers can try this at the same time, but only one can claim the current due slot.

The next_check_at = ? condition is easy to miss. It stops an old queue message from claiming a target after the schedule has moved forward.

Why the lease must expire

A Boolean processing column can get stuck forever.

Imagine the Worker claims a target, then crashes. Nothing clears the flag.

An expiring lease solves this. Another delivery can claim the same due slot after five minutes.

The lease needs a random token too. When a consumer updates the target, it includes its token in the WHERE clause:

UPDATE monitor_targets
SET lease_token = NULL, lease_expires_at = NULL
WHERE id = ? AND lease_token = ?;

This stops an old, slow consumer from clearing a newer consumer’s lease.

Set the lease duration longer than a normal job. If jobs can run for a long time, renew the lease while they run.

Check for completed work

After claiming the target, look for an existing check:

const existing = await env.DB.prepare(`
  SELECT status, attempt_count
  FROM source_checks
  WHERE target_id = ? AND scheduled_for = ?
  LIMIT 1
`)
  .bind(message.targetId, message.scheduledFor)
  .first<{ status: string; attempt_count: number }>()

if (existing && existing.status !== 'failed') {
  return { kind: 'skipped' as const }
}

A successful row means the durable work already happened. We can safely skip the fetch.

A failed row can be reused for another attempt. Update its attempt_count instead of inserting another row.

This keeps every attempt for one due slot together.

Save the result before acknowledging

Now we can fetch the pricing page and calculate a hash:

const response = await fetch(target.url, {
  signal: AbortSignal.timeout(20_000),
})

if (!response.ok) {
  throw new Error(`Pricing page returned HTTP ${response.status}`)
}

const html = await response.text()
const bytes = new TextEncoder().encode(html)
const digest = await crypto.subtle.digest('SHA-256', bytes)
const resultHash = Array.from(new Uint8Array(digest))
  .map(byte => byte.toString(16).padStart(2, '0'))
  .join('')

Before acknowledging the message, save the result and advance the target in one D1 batch:

const [checkWrite, scheduleWrite] = await env.DB.batch([
  env.DB.prepare(`
    INSERT INTO source_checks (
      work_id,
      target_id,
      scheduled_for,
      checked_at,
      status,
      attempt_count,
      result_hash
    )
    SELECT ?, ?, ?, ?, 'unchanged', 1, ?
    FROM monitor_targets
    WHERE id = ? AND lease_token = ?
    ON CONFLICT(work_id) DO UPDATE SET
      checked_at = excluded.checked_at,
      status = excluded.status,
      attempt_count = source_checks.attempt_count + 1,
      result_hash = excluded.result_hash,
      error_message = NULL
    WHERE EXISTS (
      SELECT 1
      FROM monitor_targets
      WHERE id = ? AND lease_token = ?
    )
  `).bind(
    message.workId,
    message.targetId,
    message.scheduledFor,
    now,
    resultHash,
    message.targetId,
    lease.token,
    message.targetId,
    lease.token
  ),
  env.DB.prepare(`
    UPDATE monitor_targets
    SET
      next_check_at = ?,
      lease_token = NULL,
      lease_expires_at = NULL
    WHERE id = ? AND lease_token = ?
  `).bind(nextCheckAt, message.targetId, lease.token),
])

if (
  checkWrite.meta.changes !== 1 ||
  scheduleWrite.meta.changes !== 1
) {
  throw new Error('The target lease changed before commit')
}

D1 runs the batch as a transaction. If one statement fails, neither change is committed.

Both statements are conditional on the same lease token. Only call message.ack() after the batch succeeds and both results report one changed row.

If the Worker stops before the batch, the lease expires and the work runs again. If it stops after the batch but before the acknowledgment, the next delivery sees the completed due slot and skips it.

That’s the idempotency boundary.

Preserve the original cadence

It is tempting to schedule the next run like this:

const nextCheckAt = Date.now() + intervalMinutes * 60_000

Don’t do this.

If a daily 08:00 job runs three hours late, its new time becomes 11:00. Another delay moves it again.

Calculate from the original due slot instead:

function getNextCheckAt(
  scheduledFor: number,
  intervalMinutes: number,
  now: number
) {
  const interval = intervalMinutes * 60_000
  const elapsed = Math.max(0, now - scheduledFor)
  const intervalsPassed = Math.floor(elapsed / interval) + 1

  return scheduledFor + intervalsPassed * interval
}

Suppose a daily job was due on August 4 at 08:00. It finally runs on August 6 at 10:00.

This function schedules the next check for August 7 at 08:00. It skips missed slots and keeps the original cadence.

Retry only transient failures

Not every failure deserves a retry.

A timeout, HTTP 429, or HTTP 503 may work later. An unsupported page format or an oversized response probably won’t.

Return that decision from the job:

type CheckResult =
  | { kind: 'completed' }
  | { kind: 'skipped' }
  | { kind: 'failed'; retryable: boolean }

Then let the queue consumer choose what happens:

async function consumeMessage(message, env: Env) {
  try {
    const result = await runCheck(message.body, env, message.attempts)

    if (result.kind === 'failed' && result.retryable) {
      const delaySeconds = Math.min(
        900,
        30 * 2 ** Math.min(message.attempts - 1, 5)
      )

      message.retry({ delaySeconds })
      return
    }

    message.ack()
  } catch (error) {
    console.error(error)
    message.retry()
  }
}

This starts with a 30-second delay and grows to a maximum of 15 minutes.

For a retryable failure, keep next_check_at on the same due slot and release the lease. The retry will claim that slot again.

Put lease release in runCheck()’s finally block so unexpected exceptions follow the same rule:

let workFinalized = false

try {
  // Fetch, save the result, and advance the schedule.
  workFinalized = true
} finally {
  if (!workFinalized) {
    await env.DB.prepare(`
      UPDATE monitor_targets
      SET lease_token = NULL, lease_expires_at = NULL
      WHERE id = ? AND lease_token = ?
    `).bind(message.targetId, lease.token).run()
  }
}

Set workFinalized only after a successful result or a permanent failure has been saved and the schedule advanced. Do not set it for a retryable failure.

For a permanent failure, save it and advance to the next slot. Otherwise the cron will enqueue the same broken work forever.

Finally, process every message in the queue handler:

export default {
  async queue(batch, env, _context) {
    for (const message of batch.messages) {
      await consumeMessage(message, env)
    }
  },
} satisfies ExportedHandler<Env, CheckMessage>

After max_retries, Cloudflare moves an unprocessed message to the dead letter queue. Keep that queue configured so failed work does not disappear.

What each protection does

We added several protections because each solves a different problem:

• the due slot gives the scheduled run a stable identity
• the deterministic work ID makes duplicate messages recognizable
• the unique D1 index protects against application bugs
• the conditional lease stops concurrent consumers
• the lease expiration recovers work after a crash
• the lease token stops an old consumer from clearing a new lease
• the D1 batch commits the result and schedule together
• acknowledging last prevents silent data loss
• calculating from the due slot preserves cadence after delays

My advice is to assume every background message can run twice.

Once you make that assumption, retries stop being scary. They become a normal part of the design.