Update examples and benchmarks for downstream runtime

Author: kanutocd

CHANGELOG.md

@@ -3,6 +3,8 @@
 ## 0.2.0 Unreleased
 
 - Optional downstream processor delegation to `cdc-concurrent` and `cdc-parallel`.
+- Rails example now demonstrates `cdc-concurrent` downstream processing.
+- Benchmark can measure direct downstream delegation overhead.
 
 ## 0.1.2
 

README.md

@@ -134,6 +134,8 @@ bundle exec rake benchmark:enqueue
 ```
 
 Set `CDC_SOLID_QUEUE_BENCH_EVENTS` to control the event count.
+Set `CDC_SOLID_QUEUE_BENCH_MODE=downstream_direct` to measure direct downstream
+processor delegation overhead without Solid Queue enqueue translation.
 
 Example local result on Ruby 3.4.9:
 

benchmark/enqueue_benchmark.rb

@@ -2,6 +2,7 @@
 
 require 'benchmark'
 require 'cdc/solid_queue'
+require 'cdc/core'
 
 # Fake Active Job class used by the enqueue benchmark.
 class BenchmarkJob
@@ -16,7 +17,14 @@ def perform_later(_payload)
   end
 end
 
+BenchmarkProcessor = Class.new(CDC::Core::Processor) do
+  def process(event)
+    event
+  end
+end
+
 events = Integer(ENV.fetch('CDC_SOLID_QUEUE_BENCH_EVENTS', '10000'))
+mode = ENV.fetch('CDC_SOLID_QUEUE_BENCH_MODE', 'enqueue')
 
 config = CDC::SolidQueue::Configuration.new
 config.processor_job = BenchmarkJob
@@ -33,9 +41,26 @@ def perform_later(_payload)
 }
 
 elapsed = Benchmark.realtime do
-  events.times { enqueuer.enqueue(event) }
+  case mode
+  when 'enqueue'
+    events.times { enqueuer.enqueue(event) }
+  when 'downstream_direct'
+    config.downstream_processor = BenchmarkProcessor.new
+    config.downstream_runtime = :direct
+    processor = CDC::SolidQueue::DownstreamProcessor.new(config)
+    change_event = CDC::SolidQueue::EventSerializer.load_event(event)
+    events.times { processor.process(change_event) }
+  else
+    raise ArgumentError, "unknown benchmark mode: #{mode}"
+  end
 end
 
 rate = events / elapsed
 
-puts format('events=%<events>d elapsed=%<elapsed>.4fs rate=%<rate>.2f events/s', events:, elapsed:, rate:)
+puts format(
+  'mode=%<mode>s events=%<events>d elapsed=%<elapsed>.4fs rate=%<rate>.2f events/s',
+  mode:,
+  events:,
+  elapsed:,
+  rate:
+)

examples/rails_app/.env.example

@@ -2,3 +2,5 @@ DATABASE_URL=postgres://postgres:postgres@127.0.0.1:55432/cdc_solid_queue_exampl
 CDC_SLOT=cdc_solid_queue_example
 CDC_PUBLICATION=cdc_publication
 CDC_AUTO_CREATE_SLOT=true
+CDC_CONCURRENT_CONCURRENCY=100
+CDC_CONCURRENT_TIMEOUT=5.0

examples/rails_app/Gemfile

@@ -2,6 +2,7 @@
 
 source 'https://rubygems.org'
 
+gem 'cdc-concurrent', path: '../../../cdc-concurrent'
 gem 'cdc-solid-queue', path: '../..'
 gem 'rails', '~> 7.1'
 gem 'solid_queue', '~> 1.0'

examples/rails_app/README.md

@@ -13,7 +13,8 @@ the gem repository does not need to vendor a full app skeleton.
 - `config/database.yml` points Rails at the local PostgreSQL container.
 - `config/initializers/cdc_solid_queue.rb` configures ingestion.
 - `app/jobs/application_job.rb` provides the Rails job base class.
-- `app/jobs/user_changed_job.rb` consumes `CDC::Core::ChangeEvent` objects.
+- `app/jobs/user_changed_job.rb` hands events to the configured downstream runtime.
+- `app/processors/webhook_processor.rb` is a `cdc-concurrent` processor.
 - `compose.yml` starts PostgreSQL with logical replication enabled.
 - `db/init/01_cdc_example.sql` creates the example table and publication.
 
@@ -40,8 +41,8 @@ bin/rails cdc_solid_queue:start
 ```
 
 The task streams pgoutput payloads, normalizes them to `CDC::Core::ChangeEvent`,
-enqueues the configured job through Solid Queue, and advances the checkpoint
-after enqueue succeeds.
+enqueues the configured job through Solid Queue, executes `WebhookProcessor`
+through `cdc-concurrent`, and advances the checkpoint after enqueue succeeds.
 
 Generate a change event from another terminal:
 

examples/rails_app/app/jobs/user_changed_job.rb

@@ -1,12 +1,6 @@
 # frozen_string_literal: true
 
-# Example job that consumes normalized CDC events.
+# Example job that hands normalized CDC events to configured downstream runtimes.
 class UserChangedJob < ApplicationJob
   include CDC::SolidQueue::ProcessorJob
-
-  def process(event)
-    Rails.logger.info(
-      "cdc event #{event.operation} #{event.qualified_table_name} primary_key=#{event.primary_key.inspect}"
-    )
-  end
 end

examples/rails_app/app/processors/webhook_processor.rb

@@ -0,0 +1,16 @@
+# frozen_string_literal: true
+
+require 'cdc/core'
+require 'cdc/concurrent'
+
+# Example I/O-oriented CDC processor executed through cdc-concurrent.
+class WebhookProcessor < CDC::Core::Processor
+  concurrent_safe!
+
+  def process(event)
+    Rails.logger.info(
+      "downstream cdc event #{event.operation} #{event.qualified_table_name} primary_key=#{event.primary_key.inspect}"
+    )
+    CDC::Core::ProcessorResult.success(event)
+  end
+end

examples/rails_app/config/initializers/cdc_solid_queue.rb

@@ -6,6 +6,12 @@
   config.preserve_order = true
   config.ordering_key = :primary_key
   config.checkpoint = CDC::SolidQueue::Checkpoint.new
+  config.downstream_processor = WebhookProcessor.new
+  config.downstream_runtime = :concurrent
+  config.downstream_options = {
+    concurrency: Integer(ENV.fetch('CDC_CONCURRENT_CONCURRENCY', '100')),
+    timeout: Float(ENV.fetch('CDC_CONCURRENT_TIMEOUT', '5.0'))
+  }
   config.postgresql = {
     database_url: ENV.fetch('DATABASE_URL'),
     slot: ENV.fetch('CDC_SLOT', 'cdc_solid_queue'),