Distributed job scheduler: the design decisions in depth

This is the second part of the distributed job scheduler. The first one is the tour; this one takes each decision and shows the alternatives I rejected, the queries that implement it and the edge cases it has to survive. It assumes you know the two words from part 1: a task is the configuration, a job is one execution of it.

The jobs table is the queue#

The obvious design puts a broker between the scheduler and the executors: the scheduler publishes jobs to SQS, RabbitMQ or Kafka, and executors consume them. I rejected it because of the dual write: the job has to be in the database and in the queue, and no transaction spans both. If the scheduler commits and dies before publishing, the job never runs. If it publishes first and the commit fails, an executor runs a job the database has never heard of.

Kafka had a second problem for this workload. A consumer commits one offset per partition, meaning “everything before this is done”, so a job that takes 30 seconds holds back every job behind it in that partition.

The fair alternative is the transactional outbox: write the job and an outbox row in the same transaction, and let a relay publish the outbox to the queue. That removes the dual write, but adds a relay to run and monitor and some latency. If the database is already the source of truth and can hand out work safely, the queue adds nothing.

That last condition is what SKIP LOCKED gives. I took the idea from Shopify moving inventory reservations from Redis to MySQL: they could not update two stores atomically, so they kept one, made each unit a row, and let concurrent transactions claim different rows. The executor’s claim is two statements in one short transaction:

SELECT id
FROM jobs
WHERE status = 'pending'
  AND next_attempt_at <= NOW(3) + INTERVAL ? SECOND   -- up to one poll ahead
ORDER BY priority, next_attempt_at
LIMIT ?                                               -- free workers
FOR UPDATE SKIP LOCKED;

UPDATE jobs
SET status           = 'running',
    worker_id        = ?,                             -- the pod name
    attempt          = attempt + 1,
    started_at       = NOW(3),
    lease_expires_at = GREATEST(next_attempt_at, NOW(3)) + INTERVAL (timeout_seconds + 30) SECOND,
    updated_at       = NOW(3)
WHERE id IN (...);

SKIP LOCKED passes over rows another transaction already locked instead of waiting for them, so two executors claiming at the same moment get disjoint sets. The transaction commits before any HTTP call, so a row lock lasts milliseconds and a slow endpoint never holds one.

The price is that every claim, result and retry lands on MySQL, so the jobs table needs sharding (by hash, with something like Vitess) earlier than a broker-based design would. There is also an open question I still have to measure: Shopify found that SKIP LOCKED under REPEATABLE READ takes gap locks when the range is empty, which blocked their inserts. The claim and the scheduler’s insert touch the same index range, so if it shows up under load, the claim moves to READ COMMITTED.

MySQL for the hot table, not Postgres#

A job goes through several state changes, and every one of them is a compare-and-set, an update that only applies if the row is still in the state the writer expects:

In MySQL each of those is just a WHERE. In Postgres this table would be expensive. Its indexes point to the physical location of a row version, and an update writes a new version, so it adds entries to every index unless it qualifies as a HOT update (heap-only tuple: the new version stays on the same page and the indexes are left alone). HOT is impossible when an indexed column changes, and the claim needs status indexed. And a queue table where every row is updated several times and then deleted is the worst case for vacuum and bloat.

InnoDB’s secondary indexes store the primary key instead of a physical address, so an update touches only the indexes whose columns changed. On AWS that means RDS for MySQL or Aurora MySQL, both on InnoDB.

The jobs table has one index per reader: the claim (status, priority, next_attempt_at), the reconciler (status, lease_expires_at), the archiver (status, finished_at), the DLQ (tenant_id, status), and UNIQUE (task_id, scheduled_at) so the scheduler never creates a job twice. Ids are UUIDv7, ordered by time, so inserts append to the end of the clustered index and an id works as a pagination cursor.

Cassandra for history, and moving jobs there safely#

Cassandra was the other candidate for the hot table, because it takes huge write volumes. It resolves conflicting writes with last-write-wins (the newest timestamp wins, nothing is compared), and a conditional update needs a lightweight transaction, a Paxos round between replicas. Doing that on every state change of every job would cost far more than it saves.

A finished job, though, never changes again. History is append-only, and that is exactly what Cassandra is good at. The model is close to how Discord stores messages, with a time bucket in the partition key:

CREATE TABLE job_history (
    tenant_id     uuid,
    task_id       uuid,
    bucket        int,        -- UTC month of scheduled_at, e.g. 202610
    scheduled_at  timestamp,
    job_id        uuid,
    status        text,       -- succeeded | dead
    -- attempt, http_status, error, response_body, timestamps...
    PRIMARY KEY ((tenant_id, task_id, bucket), scheduled_at, job_id)
) WITH CLUSTERING ORDER BY (scheduled_at DESC, job_id DESC)
  AND default_time_to_live = 2592000;  -- 30 days

Without the monthly bucket, a cron that runs every minute would grow one partition forever. Rows are clustered newest first, the order the history API returns, and the TTL deletes them after 30 days with no cleanup job. A second table, job_by_id, keyed by (tenant_id, job_id), says where each job lives, because Cassandra can only read by the full partition key. The tenant is part of both partition keys, so a read with the wrong tenant finds nothing: isolation does not depend on remembering a filter.

On AWS the same model fits DynamoDB (partition key tenant#task#month, sort key scheduled_at#job_id, native TTL) or Amazon Keyspaces, which speaks the Cassandra protocol. The model stays; who operates it changes.

The archiver is the only dual write left, and the order of its writes decides what a crash leaves behind. It writes to Cassandra first and deletes from MySQL second, so a crash in between leaves the job in both stores, which the job service handles (MySQL wins, duplicates are dropped by id). The other order could delete a job that never reached the history. Writing twice is harmless, because a Cassandra insert is an upsert by primary key. The delete checks the job as it was read, so a dead job retried by hand in the meantime no longer matches and stays in the queue:

DELETE FROM jobs WHERE (id, status, attempt) IN ((?, ?, ?), ...);

Inserts run in parallel, at most 64 at a time, instead of a BATCH, because a batch across partitions goes through one coordinator and its batch log. Empty columns are left unset instead of written as null, because a null in an insert writes a tombstone, a deletion marker that slows reads until compaction. Succeeded jobs leave MySQL after an hour, dead ones after 14 days, the window in which the DLQ can retry them.

Getting jobs out on time#

The scheduler runs one transaction per tick:

SELECT ... FROM tasks
WHERE status = 'active' AND next_run_at <= ?   -- now + lookahead
ORDER BY next_run_at
LIMIT 1000
FOR UPDATE;

Then it inserts all the jobs in one statement with ON DUPLICATE KEY UPDATE id = id, advances every next_run_at in one more, and commits: either both happened or neither did. The unique key on (task_id, scheduled_at) turns a repeated insert into a no-op, which happens when a task resumed inside the lookahead window gets a next_run_at that already has its job. That second statement used to be one UPDATE per task, an N+1 that turned out to be the bottleneck of the whole system, as the benchmarks post tells.

Four details in that loop matter more than they look:

• It uses FOR UPDATE, not SKIP LOCKED. A second scheduler waits for the first to commit, then sees the advanced next_run_at and skips those tasks. A second replica adds availability, not throughput.
• The query has no lower bound. A task the scheduler reaches late is created late, but never skipped.
• The next run is computed from the previous next_run_at, not from now. After an outage, each tick creates the next missed run of every overdue task until it catches up. A paused task that is resumed is the exception: the task service moves it to its next future run.
• It never touches version. Clients edit tasks with If-Match on that version, and bumping it on every run would break every edit of a task that runs each minute.

The other half of punctuality is in the executor. It claims jobs up to one polling interval ahead and parks each one on its own timer in memory, so it fires on the exact second without querying the database every second. The lease starts at the job’s time, GREATEST(next_attempt_at, NOW(3)), plus the timeout plus 30 seconds. I first counted it from the claim, which would let the reconciler take back jobs that were only waiting for their timer.

Claims, leases and late results#

I first planned a Redis lock next to the lease and dropped it: the claim already writes the job’s row, so the lock would be a second source of truth that can disagree with the database, and can be lost in a failover, without saving a single write. The lease lives in the row, worker_id and lease_expires_at, and a job is never executed before its claim is committed, so if the executor dies mid-call the job is still in the table.

The reconciler is the other half of the lease. Every interval it runs:

UPDATE jobs
SET status          = IF(attempt >= max_attempts, 'dead', 'pending'),
    next_attempt_at = NOW(3),
    error           = 'lease expired',
    worker_id       = NULL, lease_expires_at = NULL, ...
WHERE status = 'running' AND lease_expires_at < NOW(3)
LIMIT ?;

The claim already added the attempt, so the reconciler does not add it again and a job never runs more than max_attempts times. It goes back with next_attempt_at = NOW(3) because it is already late, and the LIMIT keeps a sweep after a long outage from locking the whole table.

The late result#

The interesting race is a result that arrives after the lease expired. Executor A claims a job with attempt = 1, gets a 200 just before the timeout, and sends its UPDATE over a slow connection. Meanwhile the lease expires, the reconciler returns the job, and executor B claims it with attempt = 2. Then A’s write arrives. Every result write carries WHERE id = ? AND status = 'running' AND worker_id = ? AND attempt = ?, and that is how the three possible timings end:

attempt is the fencing token: it grows with every claim, so a write that carries an older one is rejected. There is no race inside MySQL either. Two updates on one row are serialized by InnoDB’s exclusive row lock, and an UPDATE evaluates its WHERE against the latest committed version of the row (a current read), even under REPEATABLE READ.

What the guard cannot prevent is the endpoint being called twice: A already called it and B calls it again. That is the at-least-once contract.

Retries, the DLQ and what clients must do#

I first planned to retry only 5xx responses, and ended up with the same rule as Google Cloud Tasks: anything that is not a 2xx is retried, including timeouts and connection errors. Redirects are not followed, so a 3xx counts as a failure too. The backoff is backoff_base × 2^(attempt - 1), capped at one day. After max_attempts the job is dead, and the DLQ is simply the dead jobs of a tenant.

A manual retry from the DLQ resets the same job: same id, so the same Idempotency-Key, and same scheduled_at, so the endpoint knows which run it is. It is conditional on status = 'dead', and returns 404 if the job is gone, 409 if it is not dead and 202 otherwise.

Every attempt carries Idempotency-Key (the job id), X-Scheduled-At and X-Attempt. On the receiving side, the client guide in the repo asks the tenant to store each key with a UNIQUE constraint in the same transaction as the work, answer a repeated key with a 2xx without redoing it, and keep keys for the 14 days a dead job can still be retried.

SSRF and the API’s smaller decisions#

The task service rejects localhost, private ranges and the cloud metadata address when a task is created, but that alone fails against DNS rebinding, as part 1 explains. So the executor checks again at the only moment the real IP is known, in the dialer’s Control hook, after DNS resolution and right before connect():

func isPublic(ip netip.Addr) bool {
  ip = ip.Unmap() // ::ffff:10.0.0.1 is 10.0.0.1
  if !ip.IsGlobalUnicast() || ip.IsPrivate() {
    return false
  }
  for _, prefix := range blockedPrefixes { // CGNAT, NAT64, 6to4, reserved...
    if prefix.Contains(ip) {
      return false
    }
  }
  return true
}

Only ports 80 and 443 pass. Proxy: nil, because through a proxy the hook would only see the proxy’s address; no redirects, because a public endpoint could answer 302 to an internal URL; and a generic destination not allowed, because the tenant reads the error and the detail would map the network for them. The attack in full is in my notes.

The task API has a handful of smaller decisions, each with a reason:

• Idempotent POST /tasks. The Idempotency-Key is a column of tasks with UNIQUE (tenant_id, idempotency_key) and a SHA-256 of the body, rather than a separate table. The same key with the same body returns the original task; with another body, 422. When two requests race, the unique key makes one fail with error 1062, and it returns the winner’s task.
• Optimistic concurrency with If-Match. Every edit sends the version it read: no header is 428, a stale version is 412. next_run_at belongs to the scheduler, so an edit never writes it.
• No secrets in the task. Authorization, Cookie and credentials inside the URL are rejected; a task holds a secret_ref instead.
• UTC everywhere. Dates are stored in UTC and the connection sets the session time_zone, because the driver’s loc=UTC does not affect NOW(). Cron expressions are evaluated in each task’s time zone, so “09:00 in Madrid” follows daylight saving.

The six services, use case by use case#

Every service follows the same hexagonal layout: a domain with the rules, ports for what it needs and offers, a service with the use cases and adapters for MySQL, Cassandra, HTTP and cron parsing. Swapping a store means writing an adapter, not touching a use case.

The job service’s history merges both stores with two pointers, each sorted by scheduled_at descending, because an old dead job can still be in MySQL while newer ones are already in Cassandra. If one store fails, the request fails: a partial history would look complete.

The numbers behind several of these decisions, and the bug they exposed, are in part 3, the load tests.