[GSoC Proposal Draft] - Digvijay Rawat - SQL Adapter for Background Jobs

[Digvijay-x1]

SQL Adapter for Background Jobs

mentored by @rsamoilov & @cuneyter

Introduction

Rage is a Ruby web framework built for speed, using fibers and the Iodine HTTP server to achieve high performance. Its built-in background job system, Rage::Deferred, allows developers to offload work like sending emails or calling external APIs without adding heavyweight dependencies like Sidekiq or Redis.

This proposal introduces a SQL adapter for Rage::Deferred, enabling tasks to survive container restarts and be distributed across multiple pods in cloud-native deployments.

---

Problem Understanding

The Current Architecture

Rage's background job system (Rage::Deferred) works entirely in-process. When a task is enqueued, it's added to a fiber-based in-memory queue and executed by the same server process that handles HTTP requests.

To protect against data loss, Rage uses a Disk Backend (Rage::Deferred::Backends::Disk) as a Write-Ahead Log (WAL):

1. On enqueue — the task is serialized and appended to a local file on disk.
2. On completion — the task entry is removed from the file.
3. On server restart — pending_tasks reads the WAL file, loads all incomplete tasks back into memory, and re-executes them via __load_tasks.

The disk backend uses OS-level file locking (flock) to ensure that only one Iodine worker process owns each WAL file. This prevents duplicate execution within a single server, and periodic file rotation keeps storage manageable.

The Problem in Cloud Environments

This architecture breaks in Kubernetes and similar cloud platforms:

• Ephemeral storage: When a K8s pod restarts (crash, OOM kill, rolling update), the local filesystem is wiped. All pending tasks in the WAL are permanently lost.
• flock doesn't work across pods: Disk-level file locks are local to a single node. When two pods on different nodes run the same Rage app, they cannot coordinate through file locks.
• No task distribution: The disk backend has no mechanism for sharing work across pods. If Pod A has 500 queued tasks and Pod B has 0, there's no rebalancing.

These aren't edge cases. They are the default behavior in any containerized deployment. Any Rage application running on Kubernetes with background jobs is at risk of silent task loss.

The Solution

Replace the local WAL with a shared PostgreSQL/MySQL database that:

1. Persists tasks across container restarts via database storage.
2. Distributes work across pods using row-level locking (FOR UPDATE SKIP LOCKED).
3. Detects crashes via heartbeat-based stale worker sweeping.
4. Maintains Rage's simplicity no new dependencies beyond activerecord, no external job queues, no Redis.

The execution model stays the same. Tasks are processed in-memory by Iodine workers. The database acts purely as the durability and coordination layer.

---

Technical Approach

The implementation prioritizes simplicity to minimize edge cases and race conditions. It is broken into two core challenges:

1. Normal operation — enqueuing and completing tasks during steady-state.
2. Recovery and coordination — handling server boot, crashes, and multi-pod task distribution.

Database Schema

CREATE TABLE IF NOT EXISTS rage_active_workers (
  id                VARCHAR(255) PRIMARY KEY,  -- Worker process identifier
  worker_heartbeat  TIMESTAMP NOT NULL DEFAULT NOW(),
  created_at        TIMESTAMP NOT NULL DEFAULT NOW()
);

CREATE TABLE IF NOT EXISTS rage_deferred_tasks (
  id                     VARCHAR(255) PRIMARY KEY,
  owner_id               VARCHAR(255),          -- NULL = unclaimed, available for pickup
  serialized_task        TEXT NOT NULL,          -- Marshal.dump(task).dump
  publish_at             TIMESTAMP,             -- When the task becomes eligible (for delayed tasks)
  failed_execution_count INTEGER NOT NULL DEFAULT 0,  -- Tracks deserialization/infrastructure failures (DLQ)
  created_at             TIMESTAMP NOT NULL DEFAULT NOW()
);

-- Fast lookup of tasks by owner (used in crash recovery and graceful shutdown)
CREATE INDEX idx_rage_tasks_owner ON rage_deferred_tasks(owner_id);

-- Fast scan of unclaimed tasks eligible for claiming
CREATE INDEX idx_rage_tasks_claim ON rage_deferred_tasks(owner_id, publish_at);

Design decisions:

• Polling loop for delayed tasks (every 5s) instead of per-task timers.
  Setting an individual Iodine.run_after timer for each delayed task would create thousands of timers under load and doesn't survive restarts. A simple polling loop is stateless, crash-resilient, and handles clock skew gracefully.
• Idempotent schema creation (IF NOT EXISTS). The create_tables method is safe to call on every boot. This lets applications auto-create tables in after_initialize without migration tooling, which is critical for ephemeral containers that may be the first to touch the database.
• At-least-once delivery semantics. The architecture ensures that every enqueued task will be executed at least once, regardless of pod crashes, restarts, or scaling events. A task is deleted from the database only after successful execution; if a worker dies mid-execution, the task remains in the table and is eventually reclaimed by a surviving worker during the stale heartbeat sweep. For delayed tasks, the publish_at timestamp guarantees execution will not begin before the specified time, though it may start slightly after (within the 5-second claim polling interval). This makes the system suitable for operations where dropping a task is unacceptable (e.g., sending notifications, syncing external APIs) but where exact-once or precise timing is not required.
• owner_id is nullable. A NULL owner_id means the task is unclaimed and available for any worker to pick up. This is the central coordination signal, i.e., graceful shutdown and crash recovery both work by setting owner_id = NULL.
• No status column. The task's state is derived from the data itself rather than stored explicitly. If a task row exists with an owner_id, it is claimed by a worker (either pending or currently running we don't distinguish because both cases are handled identically). If a task exists with owner_id = NULL, it is unclaimed and waiting for a worker to pick it up. If the task row doesn't exist at all, it has been completed and deleted. This means we never need to update a status; we only INSERT on enqueue and DELETE on completion. The owner_id column alone encodes all the coordination state we need, eliminating the risk of status becoming inconsistent with reality (e.g., a task marked "running" whose worker is actually dead).
• No retry logic in the backend. The SQL backend only handles persistence and coordination. Retry logic (count, intervals, abort conditions) is handled by Rage::Deferred::Task. This keeps concerns cleanly separated.
• failed_execution_count for Dead Letter Queue. Instead of silently deleting tasks that fail to deserialize, the backend increments a failed_execution_count counter. Tasks exceeding a configurable threshold (default: 3) are automatically excluded from claims, acting as an in-table dead letter queue.

1. Normal Operation (Task Lifecycle)

Immediate tasks: When a worker enqueues a task without a delay, it inserts the task into the database with owner_id set to its own worker ID. Since Rage executes tasks in the same process that enqueues them, the task is immediately scheduled in-memory. The database row exists only to ensure durability; if the process crashes before completion, the task can be recovered.

# Simplified flow:
INSERT INTO rage_deferred_tasks (id, owner_id, serialized_task) VALUES ($1, $2, $3)  -- O(log N)
# → Task is scheduled in-memory via Iodine
# → On success: DELETE FROM rage_deferred_tasks WHERE id = $1

Delayed tasks: Tasks with a publish_at timestamp are inserted with owner_id = NULL. They sit in the database until a periodic background loop (claim_delayed_tasks, running every 5 seconds) finds eligible tasks and claims them.

Task completion: A simple DELETE by primary key:

DELETE FROM rage_deferred_tasks WHERE id = $1  -- O(log N)

Worker heartbeats: Each worker updates its worker_heartbeat timestamp every 30 seconds:

UPDATE rage_active_workers SET worker_heartbeat = NOW() WHERE id = $1  -- O(log W) ≈ O(1)

2. Server Boot and Crash Recovery

Task distribution on startup: When a worker boots (or reboots after a crash), it calls pending_tasks to claim orphaned work. To prevent thundering herd (multiple pods racing to claim all tasks), each worker limits how many tasks it claims using a configurable batch_size and FOR UPDATE SKIP LOCKED:

UPDATE rage_deferred_tasks SET owner_id = $1
WHERE id IN (
  SELECT id FROM rage_deferred_tasks
  WHERE owner_id IS NULL AND (publish_at IS NULL OR publish_at <= NOW())
  ORDER BY created_at ASC
  LIMIT $2
  FOR UPDATE SKIP LOCKED
)

This atomically claims and locks up to batch_size tasks. If two pods boot simultaneously, SKIP LOCKED ensures they claim different tasks with zero contention.

Graceful shutdown — When a pod receives SIGTERM (normal K8s scaling), each Iodine worker:

1. Sets owner_id = NULL on all its tasks (orphaning them for other workers).
2. Deletes its entry from rage_active_workers.
3. Closes the PG connection.

Other pods can immediately claim the orphaned tasks.

Hard crash — If a pod is killed without running shutdown hooks (OOM kill, node failure), the worker's heartbeat goes stale. A periodic sweeper (running every 60 seconds) on surviving workers detects stale heartbeats and reclaims tasks:

-- 1. Find stale workers (heartbeat older than threshold)
SELECT id FROM rage_active_workers
WHERE worker_heartbeat < NOW() - INTERVAL '90 seconds'

-- 2. Orphan their tasks
UPDATE rage_deferred_tasks SET owner_id = NULL WHERE owner_id = ANY($1)

-- 3. Remove stale registrations
DELETE FROM rage_active_workers WHERE id = ANY($1)

The stale_threshold (default: 90 seconds) is configurable. This introduces a recovery delay, but guarantees no duplicate execution. A task is only reclaimed after we're confident the original worker is dead.

Code Snippets

1. Initialize

  def initialize(batch_size: nil, heartbeat_interval: nil, stale_threshold: nil, max_failures: nil)
    @batch_size = batch_size || DEFAULT_BATCH_SIZE
    @heartbeat_interval = heartbeat_interval || DEFAULT_HEARTBEAT_INTERVAL
    @stale_threshold = stale_threshold || DEFAULT_STALE_THRESHOLD
    @max_failures = max_failures || DEFAULT_MAX_FAILURES

    @worker_id = "#{Socket.gethostname}-#{Process.pid}-#{SecureRandom.hex(8)}"

    # Create seed value for the task IDs
    task_id_seed = Time.now.to_i
    @task_id_base, @task_id_i = "#{task_id_seed}-#{@worker_id}", 0
    Iodine.run_every(1_000) do
      task_id_seed += 1
      @task_id_base, @task_id_i = "#{task_id_seed}-#{@worker_id}", 0
    end

    # Create tables if they don't exist
    with_connection { |conn| self.class.create_tables(conn) }

    # Register this worker
    register_worker

    # Start periodic heartbeat
    Iodine.run_every(@heartbeat_interval) { heartbeat }

    # Start periodic stale worker sweeper
    Iodine.run_every(DEFAULT_SWEEPER_INTERVAL) { sweep_stale_workers }

    # Start periodic loop to claim newly-eligible delayed tasks
    Iodine.run_every(DEFAULT_CLAIM_INTERVAL) { claim_available_tasks }
  end

2. Add

  def add(task, publish_at: nil, task_id: nil)
    persisted_task_id = task_id || generate_task_id
    serialized_task = Marshal.dump(task).dump

    with_connection do |conn|
      if publish_at
        # Delayed tasks are unclaimed — any worker can pick them up when eligible
        conn.exec_query(
          "INSERT INTO rage_deferred_tasks (id, owner_id, serialized_task, publish_at, created_at) VALUES (?, NULL, ?, ?, NOW())",
          "SQL", [[nil, persisted_task_id], [nil, serialized_task], [nil, Time.at(publish_at)]]
        )
      else
        # Immediate tasks are owned by the current worker (it will execute them in-process)
        conn.exec_query(
          "INSERT INTO rage_deferred_tasks (id, owner_id, serialized_task, created_at) VALUES (?, ?, ?, NOW())",
          "SQL", [[nil, persisted_task_id], [nil, @worker_id], [nil, serialized_task]]
        )
      end
    end

    persisted_task_id
  end

3. Remove

  def remove(task_id)
    with_connection do |conn|
      conn.exec_delete(
        "DELETE FROM rage_deferred_tasks WHERE id = ?", "SQL", [[nil, task_id]]
      )
    end
  end

4. Pending Tasks (claim_tasks)

  def pending_tasks
    claim_tasks
  end

  def claim_tasks
    with_connection do |conn|
      result = conn.exec_query(<<~SQL, "SQL", [[nil, @worker_id], [nil, @batch_size], [nil, @max_failures]])
        UPDATE rage_deferred_tasks SET owner_id = ?
        WHERE id IN (
          SELECT id FROM rage_deferred_tasks
          WHERE owner_id IS NULL
            AND (publish_at IS NULL OR publish_at <= NOW())
            AND failed_execution_count < ?
          ORDER BY created_at ASC
          LIMIT ?
          FOR UPDATE SKIP LOCKED
        )
 
      SQL

      result.rows.filter_map do |row|
        task_id, serialized_task, publish_at_epoch = row
        publish_at_epoch = publish_at_epoch&.to_i

        task = Marshal.load(serialized_task.undump)
        publish_at = publish_at_epoch == 0 ? nil : publish_at_epoch

        [task_id, task, publish_at]
      rescue StandardError => e
        # Dead Letter: increment failure count instead of deleting
        conn.exec_query(
          "UPDATE rage_deferred_tasks SET failed_execution_count = failed_execution_count + 1, owner_id = NULL WHERE id = ?",
          "SQL", [[nil, task_id]]
        )
        puts "ERROR: Can't deserialize task #{task_id} (#{e.class}): #{e.message}"
        nil
      end
    end
  end

3. Dead Letter Queue (DLQ)

The DLQ is implemented via the failed_execution_count column on the existing rage_deferred_tasks table. No separate table is needed.

1. When claim_tasks fails to deserialize a task, it increments failed_execution_count and releases the task (owner_id = NULL).
2. The claim query includes AND failed_execution_count < @max_failures (default: 5), so tasks that have failed too many times are automatically excluded.

• Separate from retry logic where retries are handled by Rage::Deferred::Task. The DLQ handles infrastructure failures.
• Auto delete after a set time.
• No cross-table transactions, no schema duplication.

Performance Analysis

All critical operations are indexed. Here is the time complexity for each:

Operation	Time Complexity	Notes
Enqueue (immediate/delayed)	O(log N)	B-Tree insert on rage_deferred_tasks
Task completion	O(log N)	PK delete
Worker heartbeat	O(log W) ≈ O(1)	PK update on small rage_active_workers table
Batch claim	O(log N + B)	Index scan on idx_rage_tasks_claim + lock B rows
Graceful shutdown	O(log N + T)	Index scan on idx_rage_tasks_owner, update T tasks
Stale worker sweep	O(W + K·T)	Full scan of small workers table, reclaim K workers' tasks

Where: N = total tasks, W = total workers, B = batch size, T = tasks per worker, K = stale workers.

Note: log₂(20,000,000) ≈ 24, so the architecture can handle very large loads

The batch claim uses idx_rage_tasks_claim(owner_id, publish_at) which makes finding unclaimed tasks O(log N) regardless of table size. The sweeper operates only on the small rage_active_workers table, keeping crash recovery fast.

API Integration

The adapter integrates into Rage's existing configuration system:

Rage.configure do
  config.deferred.backend = :sql, {
    batch_size: 150,            # Tasks to claim per sweep (default: 150)
    heartbeat_interval: 30_000, # Heartbeat interval in ms (default: 30s)
    stale_threshold: 90,        # Seconds before a worker is considered dead (default: 90s)
    max_failures: 3             # Max deserialization failures before dead-lettering (default: 3)
  }
end

The adapter uses ActiveRecord::Base.connection_pool.with_connection for all database operations.

---

Milestones & Timeline

Week	Milestone	Deliverables
1–4	Core adapter finalization	Create sql.rb using ActiveRecord for all DB operations. Address maintainer review feedback.
4–6	Testing infrastructure	Docker Compose setup for CI. Comprehensive integration tests: multi-worker claiming, delayed task promotion, crash simulation.
6–8	Configuration & documentation	Finalize config API. Write user-facing documentation: setup guide, configuration reference. Add YARD docs to all public methods.
9–10	SQLite adapter	Implement SQLite-compatible claiming strategy (BEGIN IMMEDIATE + UPDATE RETURNING).
11–12	Final polish & merge	Final code review. Update CHANGELOG. Prepare PR for upstream merge. End-to-end validation on a production-like K8s cluster.

---

Deliverables

1. Rage::Deferred::Backends::Sql — Production-ready SQL adapter using ActiveRecord, with full test coverage.
2. Configuration support — :sql backend option in Rage.configure with documented parameters.
3. Database schema — create_tables .
4. Test suite — Unit tests, integration tests, and Docker Compose CI setup.
5. Documentation — User guide and API reference.
6. E2E validation — Kubernetes deployment with multi-pod task distribution proof.

---

Validation

A working E2E test application built and deployed to Kubernetes:

• Application: A Rage app that sends Slack webhook notifications through Rage::Deferred background tasks.
• Infrastructure: 2 Rage pods + 1 PostgreSQL pod on a Kind cluster, with PVC for database durability.
• Verified: Tasks enqueue, execute across pods, and deliver Slack notifications with correct worker/host metadata.

For formal testing, a Docker Compose-based integration test suite will be developed that covers:

1. Basic enqueue/execute — Task is enqueued, executed, and removed.
2. Delayed execution — Task with delay: 10 executes after the delay.
3. Crash recovery — Kill a worker, verify its tasks are reclaimed by survivors.
4. Multi-worker distribution — Verify FOR UPDATE SKIP LOCKED distributes tasks fairly.
5. Graceful shutdown — Scale down a pod, verify tasks are orphaned and reclaimed.

About Me

I'm an undergraduate student who first started learning Ruby while preparing for my college’s technical society. The society is responsible for maintaining much of our campus infrastructure, including the ERP system, event websites, administrative portals, and other internal IT services. Since the ERP was built with Ruby on Rails, learning Ruby felt like the most natural place to start. In my second year, I began contributing to open-source projects, mostly in Ruby. Through that experience, I became interested in working on infrastructure-level problems rather than just application code. I chose the SQL adapter project for Rage because it addresses a real limitation in the framework today. Currently, when a Rage application is deployed on Kubernetes, background tasks can be lost if pods restart or crash. This makes it difficult to rely on Rage for production workloads in cloud environments. Building a durable, database-backed task backend would solve this problem by ensuring tasks persist across restarts. It would allow developers to confidently run Rage applications in serverless and containerized environments, fully leveraging the performance and speed the framework was designed for.

The problem itself is technically interesting and challenging involving distributed coordination, crash recovery, and race-condition handling. Working on it has already pushed me to explore concepts and systems that go well beyond what I’ve encountered in coursework, which is exactly the kind of learning experience I’m looking for through open-source development.

1. How much time would you be able to devote to the project?
I am having a Summer break from May to July, I will have approximately 8 weeks starting from 21st May to 19th July. During this period, I will be able to dedicate around ~40 hrs/week. (40 * 8 → 320 hrs.)
From 27th July until 15th September, which is approximately 7 weeks, I will only be able to devote around ~20 hrs/week due to my college semester being in session. (20 * 7 → 140 hrs.)

The allocated time frame is adequate for completing the project; however, I can adjust my work hours to meet the project's needs if necessary.

2. What other obligations might you need to work around during the summer?

• I am expecting my mid-semester exams to begin on 10th September, and I will need to adjust my work schedule accordingly. The exact date has not been announced by the university yet.
• Other than that, I don’t have any plans for the summer break.

3. How often, and through which channel(s), do you plan on communicating with your mentor?
I aim to provide daily updates to my mentor to maintain proper communication and ensure the project runs smoothly. For such discussions and addressing my small queries, my preferred mode of communication would be Discord.
Meetings with mentor: 2 times a week (flexible) - on Google Meet or any other platform.

“‬‭If accepted to contribute in Rage as a GSoC student this year, I decide‬ to put in my best efforts to submit quality work (code) and have regular‬ communication with my mentor and the broader community during‬ these 12 weeks. I also plan to sustain my involvement in this project‬ even after GSoC ends. Should it be that I can't make it this time, I will continue to contribute to the Rage project to the best of my‬ ability and will reapply for GSoC at it, next year.‬‭”‬

Excitedly looking forward to joining the Rage developer team this GSoC!‬

‭Thanks and Regards,‬
Digvijay Rawat ( Digvijay-x1 )‬

[rsamoilov]

Hi @Digvijay-x1

Overall, you have a very strong grasp of the core problem. I appreciate that you explicitly noted this project is about adding a data durability layer for stateless environments, rather than trying to build a distributed queue.

Your approach to handling graceful shutdowns with SIGTERM and mitigating thundering herds on boot using FOR UPDATE SKIP LOCKED is spot on.

There're a few edge cases I'd love for you to clarify or rethink for your next revision:

1. Database Connections & Fiber Concurrency

There is a bit of a contradiction in the proposal regarding Active Record. In the Solution section, you mention "no new dependencies beyond activerecord", but your code snippets use the raw pg gem, and your timeline mentions building an Active Record adapter later. We need to align on exactly which approach you are proposing as the primary deliverable.
More importantly, there is a misconception about fibers in your statement: "Rage is fiber-based and single-threaded per worker, so connection pooling is unnecessary". While a worker is single-threaded, fibers execute concurrently. If Fiber A makes a database query and yields on I/O, Fiber B might wake up and try to send a query over that exact same raw PG::Connection. This will corrupt the connection state. Whether you use Active Record or raw pg, you will need a fiber-aware connection pool. One advantage of Active Record is that such pool already exists.

2. Dead Letter Queue

Your current DB cleanup strategy relies on hard deletes upon completion, which is great for keeping the table lean. However, the proposal is currently missing a strategy for poison pills.

For example, in your pending_tasks snippet, if a task fails to deserialize via Marshal.load, the error is rescued and ignored. Because the task is never deleted or updated, it will sit in the database forever. If a pod crashes, the sweeper will orphan it, another worker will pick it up, fail to deserialize it again, and the cycle continues. Your proposal should include a plan for a Dead Letter Queue (e.g., moving failed/poisoned tasks to a separate table or marking them with an error status) so they don't clog the system.

3. SQLite Support

Your proposal relies on FOR UPDATE SKIP LOCKED, which is perfect for Postgres and MySQL. However, SQLite does not support this clause. How would you approach the pending_tasks query for a SQLite adapter to prevent contention when multiple worker processes boot simultaneously?

Looking forward to your updated version! Let me know if you have any questions about this feedback.

[Digvijay-x1]

Hello Roman,

Thank you so much for spending the time in reviewing the proposal and pointing the edge cases. I have update the proposal.

1. Database Connections & Fiber Concurrency

I have change the dependency from pg to AR, the sql code is only for references.

2. Dead Letter Queue

Added a DLQ

3. SQLite Support

I have updated the timeline for it

[rsamoilov]

Hi @Digvijay-x1

I love this! Added a couple of suggestions, but this is already very strong! Make sure to submit your proposal before March 31!

Polling loop for delayed tasks (every 5s) instead of per-task timers.

I can see the appeal here - we could use the SQL backend to remove one of the trafe-offs Rage::Deferred makes, i.e. storing all delayed tasks in memory. I'm on the fence about this approach, but if you want to go this way, I'd suggest to describe how exactly this will be accomplished. The problem here is that we want to encapsulate the DB logic in the backend class. If the backend class polls the database and schedules tasks, it creates a two-way dependency where the queue depends on the backend and the backend depends on the queue. Ideally, we'd want to avoid this and have a one-way queue -> backend dependency.

This is also very different from how Rage::Deferred operates today, so abstracting this new behaviour away while preserving how the existing disk backend works is also something you'd need to think through.

The DLQ is implemented via the failed_execution_count column on the existing rage_deferred_tasks table. No separate table is needed.

I think you do need a separate table. The limit for failed_execution_count can be changed. For example, the task has failed 5 times and is marked as dead. Then, a new code is deployed where this task class has the max_retries 10 setting, and the backend suddenly thinks the task is not dead anymore.

Create tables if they don't exist
with_connection { |conn| self.class.create_tables(conn) }

Great thinking, but this will be problematic in real-world deployments. This approach essentially hides some of the tables from the schema file. Additionally, it assumes the DB credentials given to the app process have the DDL permissions, which is not necessarily the case. If an app process only has DML permissions, this line will crash the app on boot.

Instead, there will need to be a command to generate the migrations for the SQL backend.

A periodic sweeper (running every 60 seconds) on surviving workers detects stale heartbeats and reclaims tasks

I have the same suggestion here as with the polling loop - think through the abstrations and try to avoid the two-way dependency where the backend schedules the tasks in the queue.