Files
sure/app/models/balance/materializer.rb
T
Abhinav Dhiman e3d46021c2 fix: memory leak in sidekiq (#1940)
* perf(sync): reduce per-job memory peak in Balance/Holding materialization

Profiling of SyncJob (StackProf object mode + Sidekiq memory middleware)
showed peaks of 600k-1.1M live heap slots per job and ~196k retained
ActiveModel::Attribute::FromUser objects post-GC, driven by full-history
in-memory accumulation in the balance/holding sync pipeline.

Changes:
- Replace Holding.new / Balance.new in calculators with lightweight
  Struct-based HoldingData / BalanceData. Skips AR attribute sets,
  belongs_to proxies, dirty tracking, type casting, and callbacks
  that were never used (upsert_all bypasses validations/callbacks
  anyway). Eliminates ~30% of allocations and the bulk of retained
  ActiveModel::Attribute::* instances.
- Build upsert payloads directly from struct fields instead of
  Holding/Balance#attributes.slice(...).
- Batch upsert_all in PERSIST_BATCH_SIZE (2,000) slices in both
  Balance::Materializer and Holding::Materializer so the intermediate
  attribute-hash array is bounded instead of holding the full
  multi-year history alongside the calculator output.
- Replace account.holdings.reload with account.holdings.reset in
  Holding::Materializer. Same cache invalidation, no eager re-query;
  the next consumer (Balance::SyncCache) loads on demand.
- Mutate entries in place in Balance::SyncCache#converted_entries
  instead of Entry#dup. The instances are scoped to the throwaway
  sync-cache and never persisted, so dup'ing was producing tens of
  thousands of unused FromUser/FromDatabase attribute wrappers per
  sync.

All persist paths run inside the existing Balance.transaction wrapper,
so batched upserts retain transactional atomicity. No production caller
of Balance::SyncCache or Holding::Materializer reuses the affected
instances outside the materializer's lifetime.

Test coverage: balance/{sync_cache,materializer,forward_calculator,
reverse_calculator} and holding/{materializer,forward_calculator,
reverse_calculator} plus account/syncer and sync (82 runs, 2,548
assertions, 0 failures).

* refactor(holding): stream materializer upserts to bound peak memory

Replace full-array accumulation + each_slice in Materializer#persist_holdings
with two flush-on-fill buffers (holdings_buffer_to_upsert_with_cost /
holdings_buffer_to_upsert_without_cost) that upsert and clear at
PERSIST_BATCH_SIZE, keeping peak RSS bounded to ~2x batch size.

Also add assert_not_nil guards in ReverseCalculatorTest before
dereferencing calculated.find results to surface clear failures
instead of NoMethodError.

* refactor(balance): promote BalanceData to Balance namespace and document sync mutation safety

- Extract Balance::BalanceData struct into its own file (app/models/balance/balance_data.rb)
  so it is discoverable without knowing it lived inside BaseCalculator
- Remove inline Struct definition from Balance::BaseCalculator; update build_balance
  to reference Balance::BalanceData explicitly (required because class Foo::Bar syntax
  does not nest Foo in constant lookup)
- Add comment to SyncCache#converted_entries clarifying that to_a materialises
  independent AR instances with no identity map active, making in-place mutation safe
- Update all Balance::BaseCalculator::BalanceData references in materializer_test

* test(balance): use to_h instead of attributes on BalanceData struct in waypoint test

* perf(balance,market_data): replace sort_by + first/last with minmax_by and push account-entry join to SQL

- Balance::Materializer#purge_stale_balances: replace sort_by(&:date) + first/last with minmax_by(&:date) to avoid full sort when only min/max are needed
- MarketDataImporter: replace Entry.group(:account_id).minimum(:date) (which loads all account IDs into a Hash) with a LEFT JOIN subquery that computes MIN(date) per account in SQL and exposes it as first_entry_date on the Account relation

* test(balance): use BalanceData struct in materializer purge test

* fix(holding): carry cost_basis forward onto gap-filled dates
2026-08-20 06:41:42 +02:00

119 lines
4.1 KiB
Ruby

class Balance::Materializer
# Upsert in chunks so that the intermediate attribute-hash array doesn't sit
# in memory alongside the full @balances array. Reduces peak RSS during sync
# for accounts with multi-year history.
PERSIST_BATCH_SIZE = 2_000
attr_reader :account, :strategy, :security_ids
def initialize(account, strategy:, security_ids: nil, window_start_date: nil)
@account = account
@strategy = strategy
@security_ids = security_ids
@window_start_date = window_start_date
end
def materialize_balances
Balance.transaction do
materialize_holdings
calculate_balances
Rails.logger.info("Persisting #{@balances.size} balances")
persist_balances
purge_stale_balances
if strategy == :forward
update_account_info
end
end
end
private
def materialize_holdings
@holdings = Holding::Materializer.new(account, strategy: strategy, security_ids: security_ids).materialize_holdings
end
def update_account_info
# Query fresh balance from DB to get generated column values
current_balance = account.balances
.where(currency: account.currency)
.order(date: :desc)
.first
if current_balance
calculated_balance = current_balance.end_balance
calculated_cash_balance = current_balance.end_cash_balance
else
# Fallback if no balance exists
calculated_balance = 0
calculated_cash_balance = 0
end
Rails.logger.info("Balance update: cash=#{calculated_cash_balance}, total=#{calculated_balance}")
account.update!(
balance: calculated_balance,
cash_balance: calculated_cash_balance
)
end
def calculate_balances
@balances = calculator.calculate
end
def persist_balances
current_time = Time.now
@balances.each_slice(PERSIST_BATCH_SIZE) do |slice|
account.balances.upsert_all(
slice.map { |b| b.to_h.except(:account).transform_keys(&:to_s).merge("updated_at" => current_time) },
unique_by: %i[account_id date currency]
)
end
end
def purge_stale_balances
if @balances.empty?
# In incremental forward-sync, even when no balances were calculated for the window
# (e.g. window_start_date is beyond the last entry), purge stale tail records that
# now fall beyond the prior-balance boundary so orphaned future rows are cleaned up.
if strategy == :forward && calculator.incremental? && calculator.calculation_start_date <= @window_start_date - 1
deleted_count = account.balances.delete_by(
"date < ? OR date > ?",
calculator.calculation_start_date,
@window_start_date - 1
)
Rails.logger.info("Purged #{deleted_count} stale balances") if deleted_count > 0
end
return
end
oldest_balance, newest_balance = @balances.minmax_by(&:date)
newest_calculated_balance_date = newest_balance.date
# In incremental forward-sync mode the calculator only recalculates from
# window_start_date onward, so balances before that date are still valid.
# Use calculation_start_date as the lower purge bound to preserve them —
# this is the same lower bound the calculator uses, so pre-anchor balances
# (from entries dated before the opening anchor) are not deleted.
# We ask the calculator whether it actually ran incrementally — it may have
# fallen back to a full recalculation, in which case we use the normal bound.
oldest_valid_date = if strategy == :forward && calculator.incremental?
calculator.calculation_start_date
else
oldest_balance.date
end
deleted_count = account.balances.delete_by("date < ? OR date > ?", oldest_valid_date, newest_calculated_balance_date)
Rails.logger.info("Purged #{deleted_count} stale balances") if deleted_count > 0
end
def calculator
@calculator ||= if strategy == :reverse
Balance::ReverseCalculator.new(account)
else
Balance::ForwardCalculator.new(account, window_start_date: @window_start_date)
end
end
end