Merge branch 'feat/DEX-2721/configure-BaseDeleteStaleItemsJob' into 'master'

[DEX-2721] feat: add options for configure BaseDeleteStaleItemsJob

Closes DEX-2721

See merge request nstmrt/rubygems/outbox!112

Author: Arlantir

Appraisals

@@ -19,6 +19,7 @@ versions_map.each do |rails_version, ruby_versions|
 
     appraise "rails-#{rails_version}" do
       gem "rails", "~> #{rails_version}.0"
+      gem "concurrent-ruby", "1.3.4" if rails_version.to_f < 7.1
     end
   end
 end

CHANGELOG.md

@@ -13,6 +13,16 @@ and this project adheres to [Semantic Versioning](http://semver.org/).
 
 ### Fixed
 
+## [6.14.0] - 2025-01-20
+
+### Added
+
+- Added options for configuring jobs to remove old items:
+  - `deletion_batch_size` - default 1_000
+  - `deletion_sleep_time` - pauses between `batch_size`
+  - `min_retention_period` - for items with statuses: `failed` and `discarded`
+  - `delivered_min_retention_period` - for items with statuses: `delivered`
+
 ## [6.13.1] - 2025-01-15
 
 ### Fixed

README.md

@@ -267,8 +267,12 @@ default: &default
   outbox_items: # outbox items section
     my_outbox_item: # underscored model class name
       owner: my_outbox_item_team # optional, used in Yabeda metrics
-      retention: P1W # for statuses: failed and discarded, retention period, https://en.wikipedia.org/wiki/ISO_8601#Durations
-      retention_delivered_items: PT6H # for statuses: delivered, retention period for delivered items, https://en.wikipedia.org/wiki/ISO_8601#Durations
+      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
       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
@@ -343,8 +347,12 @@ end
 inbox_items: # inbox items section
   my_inbox_item: # underscored model class name
     owner: my_inbox_item_team # optional, used in Yabeda metrics
-    retention: P1W # for statuses: failed and discarded, retention period, https://en.wikipedia.org/wiki/ISO_8601#Durations
-    retention_delivered_items: PT6H # for statuses: delivered, retention period for delivered items, https://en.wikipedia.org/wiki/ISO_8601#Durations
+    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
     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

app/jobs/sbmt/outbox/base_delete_stale_items_job.rb

@@ -5,10 +5,7 @@
 module Sbmt
   module Outbox
     class BaseDeleteStaleItemsJob < Outbox.active_job_base_class
-      MIN_RETENTION_PERIOD = 1.day
       LOCK_TTL = 10_800_000
-      BATCH_SIZE = 1_000
-      SLEEP_TIME = 0.5
 
       class << self
         def enqueue
@@ -45,7 +42,7 @@ def perform(item_class_name)
             duration_failed = item_class.config.retention
             duration_delivered = item_class.config.retention_delivered_items
 
-            validate_retention!(duration_failed)
+            validate_retention!(duration_delivered, duration_failed)
 
             logger.with_tags(box_type: box_type, box_name: box_name) do
               delete_stale_items(Time.current - duration_failed, Time.current - duration_delivered)
@@ -60,10 +57,22 @@ def perform(item_class_name)
 
       private
 
-      def validate_retention!(duration_failed)
-        return if duration_failed >= MIN_RETENTION_PERIOD
+      def validate_retention!(duration_delivered, duration_failed)
+        validate_retention_for!(
+          duration: duration_delivered,
+          min_period: item_class.config.delivered_min_retention_period,
+          error_message: "Retention period for #{box_name} must be longer than #{item_class.config.delivered_min_retention_period.inspect}"
+        )
+
+        validate_retention_for!(
+          duration: duration_failed,
+          min_period: item_class.config.min_retention_period,
+          error_message: "Retention period for #{box_name} must be longer than #{item_class.config.min_retention_period.inspect}"
+        )
+      end
 
-        raise "Retention period for #{box_name} must be longer than #{MIN_RETENTION_PERIOD.inspect}"
+      def validate_retention_for!(duration:, min_period:, error_message:)
+        raise error_message if duration < min_period
       end
 
       def delete_stale_items(waterline_failed, waterline_delivered)
@@ -111,7 +120,7 @@ def delete_items_in_batches(table, condition)
         subquery = table
           .project(table[:id])
           .where(condition)
-          .take(BATCH_SIZE)
+          .take(item_class.config.deletion_batch_size)
 
         delete_statement = Arel::Nodes::DeleteStatement.new
         delete_statement.relation = table
@@ -131,7 +140,7 @@ def delete_items_in_batches(table, condition)
           logger.log_info("Deleted #{deleted_count} #{box_type} items for #{box_name} items")
           break if deleted_count == 0
           lock_timer.checkpoint!
-          sleep(SLEEP_TIME)
+          sleep(item_class.config.deletion_sleep_time)
         end
       end
 
@@ -167,15 +176,15 @@ def delete_items_in_batches_mysql(query)
 
         loop do
           track_deleted_latency do
-            deleted_count = query.limit(BATCH_SIZE).delete_all
+            deleted_count = query.limit(item_class.config.deletion_batch_size).delete_all
           end
 
           track_deleted_counter(deleted_count)
 
           logger.log_info("Deleted #{deleted_count} #{box_type} items for #{box_name} items")
           break if deleted_count == 0
           lock_timer.checkpoint!
-          sleep(SLEEP_TIME)
+          sleep(item_class.config.deletion_sleep_time)
         end
       end
 

app/models/sbmt/outbox/base_item_config.rb

@@ -44,6 +44,22 @@ def retention_delivered_items
         end
       end
 
+      def deletion_batch_size
+        @deletion_batch_size ||= (options[:deletion_batch_size] || 1_000).to_i
+      end
+
+      def deletion_sleep_time
+        @deletion_sleep_time ||= (options[:deletion_sleep_time] || 0.5).to_f
+      end
+
+      def min_retention_period
+        @min_retention_period ||= ActiveSupport::Duration.parse(options[:min_retention_period] || "P1D")
+      end
+
+      def delivered_min_retention_period
+        @delivered_min_retention_period ||= ActiveSupport::Duration.parse(options[:delivered_min_retention_period] || "PT1H")
+      end
+
       def max_retries
         @max_retries ||= (options[:max_retries] || 0).to_i
       end

lib/generators/outbox/install/templates/outbox.yml

@@ -17,6 +17,11 @@ default: &default
 #      partition_size: 2
 #      partition_strategy: number
 #      retention: P1W
+#      min_retention_period: P1D
+#      retention_delivered_items: PT6H 
+#      delivered_min_retention_period: PT1H 
+#      deletion_batch_size: 1_000
+#      deletion_sleep_time: 0.5
 #      retry_strategies:
 #        - exponential_backoff
 #        - compacted_log
@@ -32,6 +37,11 @@ default: &default
 #      partition_size: 2
 #      partition_strategy: number
 #      retention: P1W
+#      min_retention_period: P1D
+#      retention_delivered_items: PT6H 
+#      delivered_min_retention_period: PT1H 
+#      deletion_batch_size: 1_000
+#      deletion_sleep_time: 0.5
 #      retry_strategies:
 #        - exponential_backoff
 #      transports:

lib/sbmt/outbox/version.rb

@@ -2,6 +2,6 @@
 
 module Sbmt
   module Outbox
-    VERSION = "6.13.1"
+    VERSION = "6.14.0"
   end
 end

spec/jobs/sbmt/outbox/base_delete_stale_items_job_spec.rb

@@ -11,35 +11,46 @@ def item_classes
     end
   end
 
-  let!(:item_delivered) { create(:outbox_item, created_at: created_at, status: 2) }
-  let!(:item_failed) { create(:outbox_item, created_at: created_at, status: 1) }
+  let!(:item_delivered) { create(:outbox_item, created_at: created_at, status: :delivered) }
+  let!(:item_failed) { create(:outbox_item, created_at: created_at, status: :failed) }
   let(:created_at) { 1.month.ago }
 
-  before do
-    stub_const("Sbmt::Outbox::BaseDeleteStaleItemsJob::BATCH_SIZE", 1)
-  end
-
   describe ".enqueue" do
     it "enqueue all item classes" do
       expect { job_class.enqueue }.to have_enqueued_job(job_class).with("OutboxItem")
     end
   end
 
-  it "deletes items with status 2 and old items with status 1 and tracks metrics" do
-    expect { job_class.perform_now("OutboxItem") }
-      .to change(OutboxItem, :count).by(-2)
-      .and increment_yabeda_counter(Yabeda.outbox.deleted_counter).with_tags(box_name: "outbox_item", box_type: :outbox).by(2)
-      .and measure_yabeda_histogram(Yabeda.outbox.delete_latency).with_tags(box_name: "outbox_item", box_type: :outbox)
-  end
+  describe "#perform" do
+    context "when all items exceed retention periods" do
+      it "deletes items and tracks metrics" do
+        expect { job_class.perform_now("OutboxItem") }
+          .to change(OutboxItem, :count).by(-2)
+          .and increment_yabeda_counter(Yabeda.outbox.deleted_counter).with_tags(box_name: "outbox_item", box_type: :outbox).by(2)
+          .and measure_yabeda_histogram(Yabeda.outbox.delete_latency).with_tags(box_name: "outbox_item", box_type: :outbox)
+      end
+    end
 
-  context "when an element with status 1 does not retention" do
-    let(:created_at) { 6.hours.ago }
+    context "when items do not exceed the minimum retention period" do
+      let(:created_at) { 6.hours.ago }
 
-    it "doesn't delete item with status 1 but deletes item with status 2 and tracks metrics" do
-      expect { job_class.perform_now("OutboxItem") }
-        .to change(OutboxItem, :count).by(-1)
-        .and increment_yabeda_counter(Yabeda.outbox.deleted_counter).with_tags(box_name: "outbox_item", box_type: :outbox).by(1)
-        .and measure_yabeda_histogram(Yabeda.outbox.delete_latency).with_tags(box_name: "outbox_item", box_type: :outbox)
+      it "does not delete items below the retention period but deletes others and tracks metrics" do
+        expect { job_class.perform_now("OutboxItem") }
+          .to change(OutboxItem, :count).by(-1)
+          .and increment_yabeda_counter(Yabeda.outbox.deleted_counter).with_tags(box_name: "outbox_item", box_type: :outbox).by(1)
+          .and measure_yabeda_histogram(Yabeda.outbox.delete_latency).with_tags(box_name: "outbox_item", box_type: :outbox)
+      end
+    end
+
+    context "when retention period is invalid" do
+      before do
+        allow(OutboxItem.config).to receive_messages(retention: 6.hours, min_retention_period: 1.day)
+      end
+
+      it "raises an error" do
+        expect { job_class.perform_now("OutboxItem") }
+          .to raise_error("Retention period for outbox_item must be longer than 1 day")
+      end
     end
   end
 end