Module: ScatterGather

Extended by:
ActiveSupport::Concern
Included in:
GatherJob
Defined in:
lib/scatter_gather.rb,
lib/scatter_gather/version.rb,
lib/generators/scatter_gather/install_generator.rb

Overview

Scatter-Gather Pattern for ActiveJob

This module provides a scatter-gather pattern for coordinating job execution. Jobs can wait for other jobs to complete before executing, with configurable polling, retry, and timeout behavior.

Example workflow:

# Start some scatter jobs
email_parser_job = EmailParserJob.perform_later(email_id: 123)
attachment_processor_job = AttachmentProcessorJob.perform_later(email_id: 123)
ai_categorizer_job = AICategorizerJob.perform_later(email_id: 123)

# Create a gather job that waits for all dependencies to complete
NotifyCompleteJob.gather(email_parser_job, attachment_processor_job, ai_categorizer_job).perform_later

The gather job will:

  • Check if all dependencies are complete
  • If complete: enqueue the target job immediately
  • If not complete: poll every 2 seconds (configurable), re-enqueuing itself
  • After 10 attempts (configurable): discard with error reporting

Configuration options:

- max_attempts: Number of polling attempts before giving up (default: 10)
- poll_interval: Time between polling attempts (default: 2.seconds)

Example with custom configuration:

TouchingJob.gather(jobs, poll_interval: 0.2.seconds, max_attempts: 4).perform_later(final_path)

Defined Under Namespace

Modules: Generators Classes: Completion, DependencyStatus, DependencyTimeoutError, GatherJob, GatherJobProxy

Constant Summary collapse

DEFAULT_GATHER_CONFIG =

Default configuration for gather jobs

{
  max_attempts: 10,
  poll_interval: 2
}.freeze
VERSION = 
"0.1.2"

Instance Method Summary collapse

Instance Method Details

#register_completion_for_gathering ⇒ Object

Updates the completions table with the status of this job



80
81
82
83
84
85
 
# File 'lib/scatter_gather.rb', line 80

def register_completion_for_gathering
  n_updated = ScatterGather::Completion.where(active_job_id: job_id).update_all(status: "completed", updated_at: Time.current)
  if n_updated > 0
    logger.tagged("ScatterGather").info { "Registered completion of #{self.class.name} id=#{job_id} since it will be gathered" }
  end
end