feat: Implement MongoDBBackend

[alob-mtc]

Summary

Add a MongoDBBackend implementation to support MongoDB as a storage backend for RunnerQ.

Background

With the pluggable storage abstraction layer (QueueBackend and InspectionBackend traits), RunnerQ can now support alternative backends beyond Redis. MongoDB is a natural fit for activity processing scenarios where:

• Document Model - Rich, flexible schema for activity payloads and metadata
• Atomic Operations - findOneAndUpdate for safe lease-based dequeuing at the document level
• Change Streams - Real-time event streaming for observability (requires replica set) - polling can also be considered
• Scalability - Replica sets and sharding for high availability
• Enterprise Adoption - Many organizations already have MongoDB infrastructure
• TTL Indexes - Native support for automatic expiration of results and idempotency keys

Proposed Implementation

File Structure

src/backend/
├── mod.rs
├── traits.rs
├── error.rs
├── redis/
│   └── ...
└── mongodb/
    ├── mod.rs
    ├── queue.rs
    ├── inspection.rs
    └── pool.rs

API

use runner_q::backend::MongoDBBackend;

let backend = MongoDBBackend::builder()
    .connection_string("mongodb://localhost:27017")
    .database("runnerq")
    .queue_name("my_app")
    .build()
    .await?;

let engine = WorkerEngine::builder()
    .backend(Arc::new(backend))
    .max_workers(8)
    .build()
    .await?;

Collection Structure

runnerq_{queue_name}_activities    # Main activity storage with status field
runnerq_{queue_name}_dead_letter   # Failed activities that exceeded retries
runnerq_{queue_name}_results       # Activity results (TTL: 24 hours)
runnerq_{queue_name}_idempotency   # Idempotency keys (TTL: 24 hours)
runnerq_{queue_name}_events        # Activity lifecycle events (capped or TTL)

Design Considerations

QueueBackend Implementation

Method	MongoDB Approach
enqueue()	Insert document with status: "pending" or status: "scheduled"
dequeue()	findOneAndUpdate with status: "pending" → status: "processing", set lease_deadline
ack_success()	Update status: "completed", clear lease fields
ack_failure()	Increment retry_count, set status: "scheduled" with backoff, or move to DLQ
process_scheduled()	Find status: "scheduled" where scheduled_at <= now, update to status: "pending"
requeue_expired()	Find status: "processing" where lease_deadline < now, update to status: "pending"
extend_lease()	Update lease_deadline where _id matches
store_result()	Upsert to results collection with TTL
get_result()	Find by _id in results collection
check_idempotency()	Atomic upsert with $setOnInsert, check if insert or existing

InspectionBackend Implementation

Method	MongoDB Approach
stats()	Aggregation pipeline with $group by status and priority
list_pending()	Find status: "pending", sort by priority desc, created_at asc, with skip/limit
list_processing()	Find status: "processing" with skip/limit
list_scheduled()	Find status: "scheduled", sort by scheduled_at
list_completed()	Find status: "completed", sort by completed_at desc
list_dead_letter()	Find all from dead_letter collection
get_activity()	Find by _id
get_activity_events()	Find by activity_id in events collection, limit
event_stream()	MongoDB Change Streams on activities collection

Activity Document Schema

{
  _id: UUID,
  activity_type: String,
  payload: Document,
  priority: Number,           // 4=Critical, 3=High, 2=Normal, 1=Low
  status: String,             // "pending", "processing", "scheduled", "completed", "failed", "dead_letter"
  created_at: DateTime,
  scheduled_at: DateTime?,
  started_at: DateTime?,
  completed_at: DateTime?,
  worker_id: String?,
  lease_id: String?,
  lease_deadline: DateTime?,
  retry_count: Number,
  max_retries: Number,
  timeout_seconds: Number,
  retry_delay_seconds: Number,
  metadata: Document,
  last_error: String?,
  last_error_at: DateTime?,
  idempotency_key: String?,
  score: Number?              // For priority ordering within status
}

Indexes

// Main activity queries
{ status: 1, priority: -1, created_at: 1 }  // Dequeue (pending by priority)
{ status: 1, scheduled_at: 1 }              // Process scheduled
{ status: 1, lease_deadline: 1 }            // Requeue expired
{ idempotency_key: 1 }                      // Idempotency lookups (sparse)

// Results collection (TTL)
{ created_at: 1 }, { expireAfterSeconds: 86400 }

// Idempotency collection (TTL)  
{ created_at: 1 }, { expireAfterSeconds: 86400 }

// Events collection (TTL or capped)
{ activity_id: 1, timestamp: -1 }
{ timestamp: 1 }, { expireAfterSeconds: 604800 }  // 7 days

Important Caveats

1. Change Streams Requirement: MongoDB Change Streams require a replica set or sharded cluster. They will NOT work on a standalone MongoDB instance. This affects the event_stream() implementation.

2. TTL Index Precision: MongoDB's TTL background thread runs approximately every 60 seconds. Document deletion may be delayed by up to 60+ seconds after expiration. For time-critical expiration, consider application-level checks.

3. Priority Queue Efficiency: Unlike Redis ZSET which provides O(log N) operations, MongoDB's compound index approach requires a full query + sort. For very high-throughput queues, Redis may be more performant.

4. Multi-Document Transactions: Available since MongoDB 4.0+ but require replica set deployment. For single-document operations (our primary use case), this is not a concern.

5. Licensing: MongoDB uses the Server Side Public License (SSPL).

Tasks

• Research MongoDB Rust client (mongodb crate)
• Design collection schema and indexes
• Create src/backend/mongodb/ module structure
• Implement MongoDBBackend struct with builder pattern
• Implement QueueBackend trait
• Implement InspectionBackend trait
• Handle priority queue semantics with compound indexes
• Implement Change Streams for event_stream() (with replica set fallback/error)
• Add TTL indexes for results and idempotency
• Write integration tests (requires MongoDB replica set / Docker)
• Add examples/mongodb_example.rs
• Update README with MongoDB usage section

Feature Flags

[features]
default = ["redis"]
redis = ["bb8-redis", "redis"]
mongodb = ["mongodb"]

Dependencies

[dependencies]
mongodb = { version = "2.8", optional = true }

References

• MongoDB Job Queue Best Practices
• MongoDB Limitations
• MongoDB Atlas Best Practices

Related

• Storage abstraction layer implementation (completed)
• QueueBackend and InspectionBackend traits in src/backend/traits.rs
• Reference implementation: src/backend/redis/
• Similar issue: feat: Add KafkaBackend implementation #58 (KafkaBackend)

[coderabbitai]

📝 CodeRabbit Plan Mode

Generate an implementation plan and prompts that you can use with your favorite coding agent.

• Create Plan

Examples

• Example 1
• Example 2

---

🔗 Similar Issues

Possible Duplicates

• Support Mongodb has a storage layer #5

Related Issues

• feat: Add KafkaBackend implementation #58
• feat: Add SqlBackend implementation (PostgreSQL/MySQL/SQLite) #59
• feat: Add first-class ValkeyBackend implementation #57

🔗 Related PRs

#18 - Prevent activity loss [merged]
#26 - feat: Thread-safe atomic idempotency with OnDuplicate behaviors [merged]
#37 - Replace in-memory broadcast channel with Redis Streams for cross-process event streaming [merged]
#56 - feat: Add pluggable backend abstraction layer [merged]
#60 - feat: add PostgreSQL backend (feature-gated, in development) [closed]

👤 Suggested Assignees

• alob-mtc

---

🧪 Issue enrichment is currently in open beta.

You can configure auto-planning by selecting labels in the issue_enrichment configuration.

To disable automatic issue enrichment, add the following to your .coderabbit.yaml:

issue_enrichment:
  auto_enrich:
    enabled: false

💬 Have feedback or questions? Drop into our discord!