From e7d5924ca19e461df655418288053f6f38eb9264 Mon Sep 17 00:00:00 2001 From: Shivam Mishra Date: Mon, 16 Mar 2026 12:26:38 +0530 Subject: [PATCH] feat: make the rollup backfill atomic --- .../reporting_events/backfill_service.rb | 7 +- lib/tasks/reporting_events_rollup.rake | 18 ++++- .../reporting_events_rollup_backfill_spec.rb | 53 +++++++++++++ .../reporting_events/backfill_service_spec.rb | 76 +++++++++++++++++++ 4 files changed, 151 insertions(+), 3 deletions(-) create mode 100644 spec/lib/reporting_events_rollup_backfill_spec.rb diff --git a/app/services/reporting_events/backfill_service.rb b/app/services/reporting_events/backfill_service.rb index 0deef624d..1bc585b81 100644 --- a/app/services/reporting_events/backfill_service.rb +++ b/app/services/reporting_events/backfill_service.rb @@ -25,10 +25,13 @@ class ReportingEvents::BackfillService end def perform - delete_existing_rollups start_utc, end_utc = date_boundaries_in_utc rollup_rows = build_rollup_rows(start_utc, end_utc) - bulk_insert_rollups(rollup_rows) if rollup_rows.any? + + ReportingEventsRollup.transaction do + delete_existing_rollups + bulk_insert_rollups(rollup_rows) if rollup_rows.any? + end end private diff --git a/lib/tasks/reporting_events_rollup.rake b/lib/tasks/reporting_events_rollup.rake index 82cae898a..e21442cbb 100644 --- a/lib/tasks/reporting_events_rollup.rake +++ b/lib/tasks/reporting_events_rollup.rake @@ -187,6 +187,7 @@ class ReportingEventsRollupBackfill # rubocop:disable Metrics/ClassLength end print_success(account, days_processed, total_days, Time.current - start_time) + prompt_enable_rollup_read_path(account) rescue StandardError => e print_failure(e, days_processed, total_days) end @@ -201,13 +202,28 @@ class ReportingEventsRollupBackfill # rubocop:disable Metrics/ClassLength puts "Average per Day: #{(elapsed_time / days_processed).round(3)} seconds" puts '' puts 'Next steps:' - puts "1. Enable feature flag: Account.find(#{account.id}).enable_features!('reporting_events_rollup')" + puts '1. Verify parity before enabling the reporting_events_rollup read path.' puts '2. Verify rollups in database:' puts " ReportingEventsRollup.where(account_id: #{account.id}).count" puts '3. Test reports to compare rollup vs raw performance' puts color('=' * 70, :green) end + def prompt_enable_rollup_read_path(account) + if account.feature_enabled?(:reporting_events_rollup) + puts color('reporting_events_rollup is already enabled for this account.', :yellow, :bold) + return + end + + print 'Enable reporting_events_rollup read path now? Only do this after parity verification. (y/N): ' + confirm = $stdin.gets.to_s.chomp.downcase + puts '' + return unless %w[y yes].include?(confirm) + + account.enable_features!('reporting_events_rollup') + puts color("Enabled reporting_events_rollup for account #{account.id}", :green, :bold) + end + def print_failure(error, days_processed, total_days) puts "\n\n" puts color('=' * 70, :red) diff --git a/spec/lib/reporting_events_rollup_backfill_spec.rb b/spec/lib/reporting_events_rollup_backfill_spec.rb new file mode 100644 index 000000000..59ccd1c2a --- /dev/null +++ b/spec/lib/reporting_events_rollup_backfill_spec.rb @@ -0,0 +1,53 @@ +require 'rails_helper' +require 'rake' + +Rails.application.load_tasks unless Object.const_defined?(:ReportingEventsRollupBackfill) + +describe ReportingEventsRollupBackfill do + let(:service) { described_class.new } + let(:account) { create(:account, reporting_timezone: 'America/New_York') } + let(:date) { Date.new(2026, 2, 11) } + + describe '#execute_backfill' do + before do + allow(ReportingEvents::BackfillService).to receive(:backfill_date) + allow($stdout).to receive(:flush) + end + + it 'prompts to enable the read path only after a successful backfill' do + allow(service).to receive(:print_success) + + expect(service).to receive(:prompt_enable_rollup_read_path).with(account) + + service.send(:execute_backfill, account, date, date, 1) + end + + it 'does not prompt to enable the read path when backfill fails' do + allow(ReportingEvents::BackfillService).to receive(:backfill_date).and_raise(StandardError, 'boom') + + expect(service).not_to receive(:prompt_enable_rollup_read_path) + + expect do + service.send(:execute_backfill, account, date, date, 1) + end.to raise_error(SystemExit) + end + end + + describe '#prompt_enable_rollup_read_path' do + it 'enables the feature flag when the user confirms' do + allow($stdin).to receive(:gets).and_return("y\n") + + expect(account).to receive(:enable_features!).with('reporting_events_rollup') + + service.send(:prompt_enable_rollup_read_path, account) + end + + it 'does not enable the feature flag when the user declines' do + allow($stdin).to receive(:gets).and_return("n\n") + + expect(account).not_to receive(:enable_features!) + + service.send(:prompt_enable_rollup_read_path, account) + end + end +end diff --git a/spec/services/reporting_events/backfill_service_spec.rb b/spec/services/reporting_events/backfill_service_spec.rb index 8f0572d3b..625f9b4e7 100644 --- a/spec/services/reporting_events/backfill_service_spec.rb +++ b/spec/services/reporting_events/backfill_service_spec.rb @@ -42,6 +42,82 @@ describe ReportingEvents::BackfillService do expect(rollup.sum_value_business_hours).to eq(0) end + it 'preserves existing rollups when building replacement rows fails' do + create( + :reporting_events_rollup, + account: account, + date: date, + dimension_type: 'account', + dimension_id: account.id, + metric: 'first_response', + count: 7, + sum_value: 700, + sum_value_business_hours: 350 + ) + + service = described_class.new(account, date) + allow(service).to receive(:build_rollup_rows).and_raise(StandardError, 'boom') + + expect do + service.perform + end.to raise_error(StandardError, 'boom') + + rollup = ReportingEventsRollup.find_by!( + account_id: account.id, + date: date, + dimension_type: 'account', + dimension_id: account.id, + metric: 'first_response' + ) + + expect(rollup.count).to eq(7) + expect(rollup.sum_value).to eq(700) + expect(rollup.sum_value_business_hours).to eq(350) + end + + it 'preserves existing rollups when replacing rows fails' do + create( + :reporting_events_rollup, + account: account, + date: date, + dimension_type: 'account', + dimension_id: account.id, + metric: 'first_response', + count: 7, + sum_value: 700, + sum_value_business_hours: 350 + ) + + create_backfill_event( + name: 'first_response', + value: 100, + value_in_business_hours: 50, + user: user, + inbox: inbox, + conversation: conversation, + created_at: Time.utc(2026, 2, 11, 15) + ) + + service = described_class.new(account, date) + allow(service).to receive(:bulk_insert_rollups).and_raise(StandardError, 'boom') + + expect do + service.perform + end.to raise_error(StandardError, 'boom') + + rollup = ReportingEventsRollup.find_by!( + account_id: account.id, + date: date, + dimension_type: 'account', + dimension_id: account.id, + metric: 'first_response' + ) + + expect(rollup.count).to eq(7) + expect(rollup.sum_value).to eq(700) + expect(rollup.sum_value_business_hours).to eq(350) + end + it 'aggregates grouped rows without instantiating reporting events' do second_user = create(:user, account: account) second_inbox = create(:inbox, account: account)