Microservices often publish messages after a transaction has been committed. Writing to the database and publishing a message are two separate transactions, so they must be atomic. A failed publication of a message could lead to a critical failure of the business process.
The Outbox pattern provides a reliable solution for message publishing. The idea behind this approach is to have an "outgoing message table" in the service's database. Before the main transaction completes, a new message row is added to this table. As a result, two actions take place as part of a single transaction. An asynchronous process retrieves new rows from the database table and, if they exist, publishes the messages to the broker.
Add this line to your application's Gemfile:
gem "sbmt-outbox"
And then execute:
bundle install
Learn how to use this gem and how it works with Ruby on Rails at here https://github.com/SberMarket-Tech/outbox-example-apps
We recommend going through the configuration and files creation process using the following Rails generators:
Each generator can be run by using the --help option to learn more about the available arguments.
If you plug the gem into your application for the first time, you can generate the initial configuration:
rails g outbox:install
An ActiveRecord model can be generated for the outbox/inbox item like this:
rails g outbox:item MaybeNamespaced::SomeOutboxItem --kind outbox
rails g outbox:item MaybeNamespaced::SomeInboxItem --kind inbox
As the result, a migration and a model will be created and the outbox.yml file configured.
A transport is a class that is invoked while processing a specific outbox or inbox item. The transport must return either a boolean value or a dry monad result.
rails g outbox:transport MaybeNamespaced::SomeOutboxItem some/transport/name --kind outbox
rails g outbox:transport MaybeNamespaced::SomeInboxItem some/transport/name --kind inbox
To create an Outbox item, you should call the Interactor with the Item Model Class and event_key as arguments. The latter will be the Partitioning Key.
transaction do
some_record.save!
result = Sbmt::Outbox::CreateOutboxItem.call(
MyOutboxItem,
event_key: some_record.id,
attributes: {
payload: some_record.generate_payload,
options: {
key: some_record.id, # optional, may be used when producing to a Kafka topic
headers: {'FOO_BAR' => 'baz'} # optional, you can add custom headers
}
}
)
raise result.failure unless result.success?
end
To create multiple Outbox items in batch, you should call the Interactor with the Item Model Class and batch attributes, each item should have same list of keys. Each item should have event_key element, it will be the Partitioning Key.
transaction do
some_record.save!
another_record.save!
result = Sbmt::Outbox::CreateOutboxBatch.call(
MyOutboxItem,
batch_attributes: [
{
event_key: some_record.id,
payload: some_record.generate_payload,
options: {
key: some_record.id, # optional, may be used when producing to a Kafka topic
headers: {'FOO_BAR' => 'baz'} # optional, you can add custom headers
}
},
{
event_key: another_record.id,
payload: another_record.generate_payload,
options: {
key: another_record.id, # optional, may be used when producing to a Kafka topic
headers: {'FOO_BAR' => 'baz'} # optional, you can add custom headers
}
}
]
)
raise result.failure unless result.success?
end
We use Yabeda to collect all kind of metrics.
Example of a Grafana dashboard that you can import from a file:
OutboxfileFirst of all you should create an Outboxfile at the root of your application with the following code:
# frozen_string_literal: true
require_relative "config/environment"
# Comment out this line if you don't want to use a metrics exporter
Yabeda::Prometheus::Exporter.start_metrics_server!
config/initializers/outbox.rbThe config/initializers/outbox.rb file contains the overall general configuration.
# config/initializers/outbox.rb
Rails.application.config.outbox.tap do |config|
config.redis = {url: ENV.fetch("REDIS_URL")} # Redis is used as a coordinator service
config.paths << Rails.root.join("config/outbox.yml").to_s # optional; configuration file paths, deep merged at the application start, useful with Rails engines
# optional (worker v2: default)
config.poller = ActiveSupport::OrderedOptions.new.tap do |pc|
# max parallel threads (per box-item, globally)
pc.concurrency = 6
# max threads count (per worker process)
pc.threads_count = 1
# maximum processing time of the batch, after which the batch will be considered hung and processing will be aborted
pc.general_timeout = 60
# poll buffer consists of regular items (errors_count = 0, i.e. without any processing errors) and retryable items (errors_count > 0)
# max poll buffer size = regular_items_batch_size + retryable_items_batch_size
pc.regular_items_batch_size = 200
pc.retryable_items_batch_size = 100
# poll tactic: default is optimal for most cases: rate limit + redis job-queue size threshold
# poll tactic: aggressive is for high-intensity data: without rate limits + redis job-queue size threshold
# poll tactic: low-priority is for low-intensity data: rate limits + redis job-queue size threshold + + redis job-queue lag threshold
pc.tactic = "default"
# number of batches that one thread will process per rate interval
pc.rate_limit = 60
# rate interval in seconds
pc.rate_interval = 60
# mix / max redis job queue thresholds per box-item for default / aggressive / low-priority poll tactics
pc.min_queue_size = 10
pc.max_queue_size = 100
# min redis job queue time lag threshold per box-item for low-priority poll tactic (in seconds)
pc.min_queue_timelag = 5
# throttling delay for default / aggressive / low-priority poll tactics (in seconds)
pc.queue_delay = 0.1
end
# optional (worker v2: default)
config.processor = ActiveSupport::OrderedOptions.new.tap do |pc|
# max threads count (per worker process)
pc.threads_count = 4
# maximum processing time of the batch, after which the batch will be considered hung and processing will be aborted
pc.general_timeout = 120
# BRPOP delay (in seconds) for polling redis job queue per box-item
pc.brpop_delay = 2
end
# optional (worker v1: DEPRECATED)
config.process_items.tap do |x|
# maximum processing time of the batch, after which the batch will be considered hung and processing will be aborted
x.general_timeout = 180
# maximum batch processing time, after which the processing of the batch will be aborted in the current thread,
# and the next thread that picks up the batch will start processing from the same place
x.cutoff_timeout = 60
# batch size
x.batch_size = 200
end
# optional (worker v1: DEPRECATED)
config.worker.tap do |worker|
# number of batches that one thread will process per rate interval
worker.rate_limit = 10
# rate interval in seconds
worker.rate_interval = 60
end
end
You should create a database table in order for the process to view your outgoing messages.
create_table :my_outbox_items do |t|
t.uuid :uuid, null: false
t.string :event_name, null: false # optional, use it when you have several events per one outbox table
t.string :event_key, null: false
t.integer :bucket, null: false
t.integer :status, null: false, default: 0
t.jsonb :options
t.binary :payload, null: false # when using mysql the column type should be mediumblob
t.integer :errors_count, null: false, default: 0
t.text :error_log
t.timestamp :processed_at
t.timestamps
end
add_index :my_outbox_items, :uuid, unique: true
add_index :my_outbox_items, [:status, :id, :bucket], algorithm: :concurrently, include: [:errors_count]
add_index :my_outbox_items, [:event_name, :event_key, :id]
add_index :my_outbox_items, :created_at
You can combine various types of messages within a single table. To do this, you should include an event_name field in the table. However, this approach is only justified if it is assumed that there won't be many events, and those events will follow the same retention and retry policy.
# app/models/my_outbox_item.rb
class MyOutboxItem < Sbmt::Outbox::OutboxItem
validates :event_name, presence: true # optional
end
The outbox.yml configuration file is the main configuration for the gem, where parameters for each outbox/inbox item are located.
# config/outbox.yml
default: &default
owner: foo-team # optional, used in Yabeda metrics
bucket_size: 16 # optional, default 16, see into about the buckets at the #Concurrency section
metrics:
enabled: true # default false, yabeda server autostart with port: 9090 and path: /metrics
port: 9090 # optional, default, used in Yabeda metrics
probes:
enabled: false # optional, default true
port: 5555 # default, used for Kubernetes probes
outbox_items: # outbox items section
my_outbox_item: # underscored model class name
owner: my_outbox_item_team # optional, used in Yabeda metrics
retention: P1W #optional, default: P1W, for statuses: failed and discarded, retention period, https://en.wikipedia.org/wiki/ISO_8601#Durations
min_retention_period: P1D #optional, default: P1D, for statuses: failed and discarded, retention period, https://en.wikipedia.org/wiki/ISO_8601#Durations
retention_delivered_items: PT6H #optional, default: P1W, for statuses: delivered, retention period for delivered items, https://en.wikipedia.org/wiki/ISO_8601#Durations
delivered_min_retention_period: PT1H #optional, default: PT1H, for statuses: delivered, retention period for delivered items, https://en.wikipedia.org/wiki/ISO_8601#Durations
deletion_batch_size: 1_000 #optional, default: 1_000
deletion_sleep_time: 0.5 #optional, default: 0.5
deletion_time_window: PT4H #optional, default: PT4H, for statuses: delivered, retention period for delivered items, https://en.wikipedia.org/wiki/ISO_8601#Durations
max_retries: 3 # default 0, the number of retries before the item will be marked as failed
strict_order: false # optional, default
transports: # transports section
produce_message: # underscored transport class name
# transport reserved options
class: produce_message # optional; default is inferred from transport name
disposable: false # optional; default false; if true, the transport class will be instantiated only once
# ProduceMessage instance arguments
topic: "my-topic-name"
development:
<<: *default
test:
<<: *default
bucket_size: 2
production:
<<: *default
bucket_size: 256
CAUTION:
retry_strategies and the strict_order option at the same time# app/services/import_order.rb
class ProduceMessage
def initialize(topic:)
@topic = topic
end
def call(outbox_item, payload)
# send message to topic
true # mark message as processed
end
end
If you use Kafka as a transport, it is recommended that you use the sbmt-kafka_producer gem for this purpose.
Transports are defined as follows when event_name is used:
outbox_items:
my_outbox_item:
transports:
# transport reserved options
- class: produce_message
event_name: "order_created" # event name marker
# ProduceMessage instance arguments
topic: "order_created_topic" # some transport argument
- class: produce_message
event_name: "orders_completed"
topic: "orders_completed_topic"
The database migration will be the same as described in the Outbox pattern.
# app/models/my_inbox_item.rb
class MyInboxItem < Sbmt::Outbox::InboxItem
end
# config/outbox.yml
# see main configuration at the Outbox pattern
inbox_items: # inbox items section
my_inbox_item: # underscored model class name
owner: my_inbox_item_team # optional, used in Yabeda metrics
retention: P1W #optional, default: P1W, for statuses: failed and discarded, retention period, https://en.wikipedia.org/wiki/ISO_8601#Durations
min_retention_period: P1D #optional, default: P1D, for statuses: failed and discarded, retention period, https://en.wikipedia.org/wiki/ISO_8601#Durations
retention_delivered_items: PT6H #optional, default: P1W, for statuses: delivered, retention period for delivered items, https://en.wikipedia.org/wiki/ISO_8601#Durations
delivered_min_retention_period: PT1H #optional, default: PT1H, for statuses: delivered, retention period for delivered items, https://en.wikipedia.org/wiki/ISO_8601#Durations
deletion_batch_size: 1_000 #optional, default: 1_000
deletion_sleep_time: 0.5 #optional, default: 0.5
deletion_time_window: PT4H #optional, default: PT4H, for statuses: delivered, retention period for delivered items, https://en.wikipedia.org/wiki/ISO_8601#Durations
max_retries: 3 # default 0, the number of retries before the item will be marked as failed
transports: # transports section
import_order: # underscored transport class name
source: "kafka" # default transport arguments
# app/services/import_order.rb
class ImportOrder
def initialize(source:)
@source = source
end
def call(outbox_item, payload)
# some work to create order in the database
true # mark message as processed
end
end
If you use Kafka, it is recommended that you use the sbmt-kafka_consumer gem for this purpose.
The gem uses several types of retry strategies to repeat message processing if an error occurs. These strategies can be combined and will be executed one after the other. Each retry strategy takes one of three actions: to process the message, to skip processing the message or to skip processing and mark the message as "skipped" for future processing.
This strategy periodically attempts to resend failed messages, with increasing delays in between each attempt.
# config/outbox.yml
outbox_items:
my_outbox_item:
...
minimal_retry_interval: 10 # default: 10
maximal_retry_interval: 600 # default: 600
multiplier_retry_interval: 2 # default: 2
retry_strategies:
- exponential_backoff
This strategy ensures idempotency. In short, if a message fails and a later message with the same event_key has already been delivered, then you most likely do not want to re-deliver the first one when it is retried.
# config/outbox.yml
outbox_items:
my_outbox_item:
...
retry_strategies:
- exponential_backoff
- latest_available
The exponential backoff strategy should be used in conjunction with the latest available strategy, and it should come last to minimize the number of database queries.
Depending on which type of data is used in the event_key, it is necessary to choose the right partitioning strategy.
This strategy should be used when the event_key field contains a number. For example, it could be 52523, or some-chars-123. Any characters that aren't numbers will be removed, and only the numbers will remain. This strategy is used as a default.
# config/outbox.yml
outbox_items:
my_outbox_item:
...
partition_strategy: number
This strategy should be used when the event_key is a string or uuid.
# config/outbox.yml
outbox_items:
my_outbox_item:
...
partition_strategy: hash
rake outbox:delete_items
rake outbox:update_status_items
Example run:
rake outbox:delete_items[OutboxItem,1] # Mandatory parameters box class and status
rake outbox:update_status_items[OutboxItem,0,3] # Mandatory parameters box class, current status and new status
Both tasks have optional parameters:
- start_time # boxes are younger than the specified time, by default nil, time is specified in the format "2025-01-05T23:59:59"
- end_time # boxes are older than the specified time, by default 6.hours.ago, time is specified in the format "2025-01-05T23:59:59"
- batch_size # batch size, by default 1_000
- sleep_time # sleep time between batches, by default 0.5
Example with optional parameters:
rake outbox:delete_items[klass_name,status,start_time,end_time,batch_size,sleep_time]
rake outbox:update_status_items[klass_name,status,new_status,start_time,end_time,batch_size,sleep_time]
rake outbox:delete_items[OutboxItem,1,"2025-01-05T23:59:59","2025-01-05T00:00:00",10_000,5]
rake outbox:update_status_items[OutboxItem,0,3,"2025-01-05T23:59:59","2025-01-05T00:00:00",10_000,5]
The worker process consists of a poller and a processor, each of which has its own thread pool.
The poller is responsible for fetching messages ready for processing from the database table.
The processor, in turn, is used for their consistent processing (while preserving the order of messages and the partitioning key).
Each bunch of buckets (i.e. buckets partition) is consistently fetched by poller one at a time. Each bucket is processed one at a time by a processor.
A bucket is a number in a row in the bucket column generated by the partitioning strategy based on the event_key column when a message was committed to the database within the range of zero to bucket_size.
The number of bucket partitions, which poller uses is 6 by default. The number of poller threads is 2 by default and is not intended for customization.
The default number of processor threads is 4 and can be configured with the --concurrency option, thereby allowing you to customize message processing performance.
This architecture was designed to allow the daemons to scale without stopping the entire system in order to avoid mixing messages chronologically.
You can wrap item processing within middlewares. There are three types:
The order of execution depends on the order specified in the outbox configuration:
# config/initializers/outbox.rb
Rails.application.config.outbox.tap do |config|
config.item_process_middlewares.push(
'MyFirstItemMiddleware', # goes first
'MySecondItemMiddleware' # goes second
)
end
# config/initializers/outbox.rb
Rails.application.config.outbox.tap do |config|
config.create_item_middlewares.push(
'MyCreateItemMiddleware'
)
config.create_batch_middlewares.push(
'MyCreateBatchMiddleware'
)
end
# my_create_item_middleware.rb
class MyCreateItemMiddleware
def call(item_class, item_attributes)
# your code
yield
# your code
end
end
# my_create_batch_middleware.rb
class MyCreateBatchMiddleware
def call(item_class, batch_attributes)
# your code
yield
# your code
end
end
Example of a batch middleware:
# config/initializers/outbox.rb
Rails.application.config.outbox.tap do |config|
config.batch_process_middlewares.push(
'MyBatchMiddleware'
)
end
# my_batch_middleware.rb
class MyBatchMiddleware
def call(job)
# your code
yield
# your code
end
end
Example of an item middleware:
# config/initializers/outbox.rb
Rails.application.config.outbox.tap do |config|
config.item_process_middlewares.push(
'MyItemMiddleware'
)
end
# my_create_item_middleware.rb
class MyItemMiddleware
def call(item)
# your code
yield
# your code
end
end
Example of an polling middleware:
# config/initializers/outbox.rb
Rails.application.config.outbox.tap do |config|
config.polling_item_middlewares.push(
'MyItemMiddleware'
)
end
# my_create_polling_middleware.rb
class MyPollingItemMiddleware
def call(item)
# your code
yield
# your code
end
end
## Tracing
The gem is optionally integrated with OpenTelemetry. If your main application has `opentelemetry-*` gems, the tracing will be configured automatically.
## Web UI
Outbox comes with a [Ract web application](https://github.com/SberMarket-Tech/sbmt-outbox-ui) that can list existing outbox and inbox models.
```ruby
Rails.application.routes.draw do
mount Sbmt::Outbox::Engine => "/outbox-ui"
end
The path /outbox-ui cannot be changed for now
Under the hood it uses a React application provided as npm package.
By default, the npm packages is served from https://cdn.jsdelivr.net/npm/sbmt-outbox-ui@x.y.z/dist/assets/index.js. It could be changed by the following config option:
# config/initializers/outbox.rb
Rails.application.config.outbox.tap do |config|
config.cdn_url = "https://some-cdn-url"
end
If you want to implement some features for Outbox UI, you can serve javascript locally like the following:
npm run dev# config/initializers/outbox.rb
Rails.application.config.outbox.tap do |config|
config.ui.serve_local = true
end
We would like to see more features added to the web UI. If you have any suggestions, please feel free to submit a pull request 🤗.
| Key | Description |
|---|---|
--boxes or -b | Outbox/Inbox processors to start` |
--concurrency or -c | Number of process threads. Default 4. |
--poll-concurrency or -p | Number of poller partitions. Default 6. |
--poll-threads or -n | Number of poll threads. Default 1. |
--poll-tactic or -t | Poll tactic. Default "default". |
--worker-version or -w | Worker version. Default 2. |
| Key | Description |
|---|---|
--boxes or -b | Outbox/Inbox processors to start` |
--concurrency or -c | Number of threads. Default 10. |
dip provisiondip setupdip testSee more commands at dip.yml.
FAQs
Unknown package
We found that sbmt-outbox demonstrated a healthy version release cadence and project activity because the last version was released less than a year ago. It has 1 open source maintainer collaborating on the project.
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Research
Security News
Malicious Go packages are impersonating popular libraries to install hidden loader malware on Linux and macOS, targeting developers with obfuscated payloads.
Security News
Bybit's $1.46B hack by North Korea's Lazarus Group pushes 2025 crypto losses to $1.6B in just two months, already surpassing all of 2024's $1.49B total.
Security News
OpenSSF has published OSPS Baseline, an initiative designed to establish a minimum set of security-related best practices for open source software projects.