DOCSEARCHbyAlgolia
Submit Docs

sidekiq-rs

repository·master·Indexed 41 days ago

https://github.com/film42/sidekiq-rs

A Rust implementation of Sidekiq (rusty-sidekiq v0.15.1) compatible with the original Ruby Sidekiq for submitting and processing jobs. Built on Tokio and async by default, it provides a server and client for managing strongly typed workers, periodic cron jobs, unique jobs, and custom server-side middleware.

Tokens
11.1K
Snippets
34
Records
45
Agent score
51%

What's inside sidekiq-rs

  1. Configure unique jobs

    master
    Unique jobs prevent the same job from being enqueued multiple times within a specific timeframe. This is achieved using the unique_for(duration) option on WorkerOpts. Uniqueness is determined by a combination of the worker_name, queue_name, and a sha256_hash of the job arguments. The implementation uses a Redis SET key value NX EX duration lock.

  2. Best practice: Separate enqueue and fetch connection pools

    master

    It is recommended to use two distinct Redis connection pools: one for enqueuing jobs (pushing) and one for the Processor (fetching/consuming).

    This prevents the Processor from exhausting the pool and causing delays in the critical path of your application (like API requests) when enqueuing new jobs. It also allows you to optimize pool sizes independently for each task.

    #[tokio::main]
async fn main() -> Result<()> {
    let manager = sidekiq::RedisConnectionManager::new("redis://127.0.0.1/").unwrap();
    let redis_enqueue = bb8::Pool::builder().build(manager.clone()).await.unwrap();
    let redis_fetch = bb8::Pool::builder().build(manager).await.unwrap();

    let p = Processor::new(
        redis_fetch,
        vec!["default".to_string()],
    );
    
    // Run processor in background or separate task
    tokio::spawn(async move { p.run().await });

    // Use the enqueue pool for jobs
    ExampleWorker::perform_async(&redis_enqueue, ExampleArgs { foo: "bar".into() }).await?;

    Ok(())
}

  3. Implement a Worker with strongly typed arguments

    master

    To create a worker, define a struct for the worker itself and a separate struct for its arguments. The argument struct must implement Serialize and Deserialize. Implement the Worker<Args> trait for your worker struct. You can also define default worker options, such as the queue name, via the opts() method.

    use tracing::info;
use sidekiq::Result;

#[derive(Clone)]
struct PaymentReportWorker {}

impl PaymentReportWorker {
    fn new() -> Self {
        Self { }
    }

    async fn send_report(&self, user_guid: String) -> Result<()> {
        info!({"user_guid" = user_guid}, "Sending payment report to user");
        Ok(())
    }
}

#[derive(Deserialize, Debug, Serialize)]
struct PaymentReportArgs {
    user_guid: String,
}

#[async_trait]
impl Worker<PaymentReportArgs> for PaymentReportWorker {
    // Default worker options
    fn opts() -> sidekiq::WorkerOpts<Self> {
        sidekiq::WorkerOpts::new().queue("yolo")
    }

    // Worker implementation
    async fn perform(&self, args: PaymentReportArgs) -> Result<()> {
        self.send_report(args.user_guid).await
    }
}

  4. Schedule periodic cron jobs

    master

    Periodic jobs can be scheduled using cron strings. Use the periodic::builder to define the schedule, name, queue, and arguments, then register it with a Processor instance.

    Important: Periodic jobs are stored in Redis and are not automatically removed if you delete the code that creates them. It is a best practice to call periodic::destroy_all(redis).await? at the start of your application to clear out old schedules and ensure only currently defined jobs exist.

    // Clear out all periodic jobs and their schedules
periodic::destroy_all(redis).await?;

// Add a new periodic job
periodic::builder("0 0 8 * * *")?
    .name("Email clients with an oustanding balance daily at 8am UTC")
    .queue("reminders")
    .args(EmailReminderArgs { report_type: "outstanding_balance" })?
    .register(&mut p, EmailReminderWorker)
    .await?;

  5. Pass database connections into workers

    master

    To provide workers with access to external resources like database connections or HTTP clients, include them as fields in your worker struct. The struct must implement Clone. You can then access these resources within the perform method.

    use tracing::debug;
use sidekiq::Result;

#[derive(Clone)]
struct ExampleWorker {
    redis: RedisPool,
}

#[async_trait]
impl Worker<()> for ExampleWorker {
    async fn perform(&self, args: PaymentReportArgs) -> Result<()> {
        use redis::AsyncCommands;

        // Accessing the resource via self
        let times_called: usize = self
            .redis
            .get()
            .await?
            .unnamespaced_borrow_mut()
            .incr("example_of_accessing_the_raw_redis_connection", 1)
            .await?;

        debug!({"times_called" = times_called}, "Called this worker");
    }
}

#[tokio::main]
async fn main() -> Result<()> {
    // ...
    let mut p = Processor::new(
        redis.clone(),
        vec!["low_priority".to_string()],
    );

    p.register(ExampleWorker{ redis: redis.clone() });
}

  6. Enqueue jobs with different levels of control

    master

    You can insert jobs into Redis using several methods:

    1. Standard Enqueue: Use the worker's perform_async method with strongly typed arguments.
    2. With Overrides: Use Worker::opts() to override default settings (like the queue name) before calling perform_async.
    3. Crate-level Enqueue: Use sidekiq::perform_async for maximum control, passing the worker class name and queue name as strings.
    // 1. Standard
PaymentReportWorker::perform_async(
    &mut redis,
    PaymentReportArgs { user_guid: "USR-123".into() },
).await?;

// 2. With Overrides
PaymentReportWorker::opts()
    .queue("brolo")
    .perform_async(
        &mut redis,
        PaymentReportArgs { user_guid: "USR-123".into() },
    )
    .await?;

// 3. Crate-level control
sidekiq::perform_async(
    &mut redis,
    "PaymentReportWorker".into(),
    "yolo".into(),
    PaymentReportArgs { user_guid: "USR-123".to_string() },
).await?;

  7. Start the Sidekiq server

    master

    To run a Sidekiq server, you must:

    1. Create a RedisConnectionManager and a connection pool (e.g., using bb8).
    2. Initialize a Processor with the Redis pool and a list of queues to watch.
    3. Register your worker instances using p.register().
    4. (Optional) Add custom middlewares using p.using().
    5. Start the processing loop with p.run().
    // Redis
let manager = sidekiq::RedisConnectionManager::new("redis://127.0.0.1/").unwrap();
let mut redis = bb8::Pool::builder().build(manager).await.unwrap();

// Sidekiq server
let mut p = Processor::new(
    redis,
    vec!["yolo".to_string(), "brolo".to_string()],
);

// Add known workers
p.register(PaymentReportWorker::new());

// Custom Middlewares
p.using(FilterExpiredUsersMiddleware::new())
    .await;

// Start the server
p.run().await;

  8. Manage queue priority with BalanceStrategy

    master

    When multiple queues are provided to Processor::new, the engine must decide which queue to poll from Redis. Because the underlying Redis BRPOP command checks queues in the order they are provided, the first queue can starve others if it is always busy.

    BalanceStrategy mitigates this:

    • BalanceStrategy::RoundRobin (Default): Rotates the list of queues by 1 every time jobs are fetched, ensuring all queues have an equal opportunity to be processed.
    • BalanceStrategy::None: Does not modify the queue order. Use this only if you want strict priority based on the order of queues passed to new(), but be aware of potential starvation.

  9. Configure job retry policies

    master

    The RetryOpts enum controls how Sidekiq handles job failures. It can be converted from a bool or a usize.

    • RetryOpts::Yes: Standard Sidekiq retry behavior.
    • RetryOpts::Never: Disables retries for the job.
    • RetryOpts::Max(n): Limits retries to a specific number of attempts.

    Example conversions:

    • true -> RetryOpts::Yes
    • false -> RetryOpts::Never
    • 5 -> RetryOpts::Max(5)
    use sidekiq_rs::RetryOpts;

let opts_yes = RetryOpts::from(true);
let opts_never = RetryOpts::from(false);
let opts_limit = RetryOpts::from(5);
let opts_explicit = RetryOpts::Max(10);

  10. How RedisConnection handles namespacing

    master

    The RedisConnection maintains an internal Option<String> for the namespace.

    1. Prefixing: When a method is called, it uses namespaced_key or namespaced_keys to transform the input. If a namespace exists, it returns format!("{namespace}:{key}").
    2. Bypassing: If no namespace is set, the key is returned unchanged.
    3. Manual Control: You can use set_namespace(String) to change the namespace on an existing connection or with_namespace(String) to create a new connection instance with a namespace.

  11. Namespace Redis commands

    master

    If you are using the redis-namespace gem in Ruby, you can support namespacing in sidekiq-rs by using a connection customizer when building your bb8::Pool. This prefixes all commands used by the library with your specified namespace (e.g., my_cool_app:ZDEL ...).

    let manager = sidekiq::RedisConnectionManager::new("redis://127.0.0.1/")?;
let redis = bb8::Pool::builder()
    .connection_customizer(sidekiq::with_custom_namespace("my_cool_app".to_string()))
    .build(manager)
    .await?;

  12. Configure the number of worker tasks in `sidekiq::Processor`

    master

    The sidekiq::Processor defaults to spawning a number of worker tasks equal to the host's CPU count. For IO-bound workloads (e.g., database queries or web requests), you may want to increase this number to utilize CPU resources more efficiently. Use ProcessorConfig::num_workers() to set this value.

    #[tokio::main]
async fn main() -> Result<()> {
    // ...
    let num_workers = usize::from_str(&env::var("NUM_WORKERS").unwrap()).unwrap();
    let config: ProcessorConfig = Default::default();
    let config = config.num_workers(num_workers);
    let processor = Processor::new(redis_fetch, queues.clone())
        .with_config(config);
    // ...
}