fix: memory leak in sidekiq

app/models/balance/balance_data.rb

@@ -0,0 +1,9 @@
+Balance::BalanceData = Struct.new(
+  :account, :date, :currency, :balance, :cash_balance,
+  :start_cash_balance, :start_non_cash_balance,
+  :cash_inflows, :cash_outflows,
+  :non_cash_inflows, :non_cash_outflows,
+  :cash_adjustments, :non_cash_adjustments,
+  :net_market_flows, :flows_factor,
+  keyword_init: true
+)

app/models/balance/base_calculator.rb

@@ -118,10 +118,10 @@ def signed_entry_flows(entries)
     end
 
     def build_balance(date:, **args)
-      Balance.new(
-        account_id: account.id,
-        currency: account.currency,
+      Balance::BalanceData.new(
+        account: account,
         date: date,
+        currency: account.currency,
         balance: args[:balance],
         cash_balance: args[:cash_balance],
         start_cash_balance: args[:start_cash_balance] || 0,

app/models/balance/materializer.rb

@@ -1,4 +1,9 @@
 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)
@@ -59,24 +64,16 @@ def calculate_balances
 
     def persist_balances
       current_time = Time.now
-      account.balances.upsert_all(
-        @balances.map { |b| b.attributes
-               .slice("date", "balance", "cash_balance", "currency",
-                      "start_cash_balance", "start_non_cash_balance",
-                      "cash_inflows", "cash_outflows",
-                      "non_cash_inflows", "non_cash_outflows",
-                      "net_market_flows",
-                      "cash_adjustments", "non_cash_adjustments",
-                      "flows_factor")
-               .merge("updated_at" => current_time) },
-        unique_by: %i[account_id date currency]
-      )
+      @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
-      sorted_balances = @balances.sort_by(&:date)
-
-      if sorted_balances.empty?
+      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.
@@ -91,7 +88,8 @@ def purge_stale_balances
         return
       end
 
-      newest_calculated_balance_date = sorted_balances.last.date
+      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.
@@ -101,7 +99,7 @@ def purge_stale_balances
       oldest_valid_date = if strategy == :forward && calculator.incremental?
         account.opening_anchor_date
       else
-        sorted_balances.first.date
+        oldest_balance.date
       end
 
       deleted_count = account.balances.delete_by("date < ? OR date > ?", oldest_valid_date, newest_calculated_balance_date)

app/models/balance/sync_cache.rb

@@ -35,19 +35,23 @@ def holdings_value_by_date
 
     def converted_entries
       @converted_entries ||= account.entries.excluding_split_parents.includes(:entryable).order(:date).to_a.map do |e|
-        converted_entry = e.dup
-
         custom_rate = e.entryable.exchange_rate if e.entryable.respond_to?(:exchange_rate)
 
-        # Use Money#exchange_to with custom rate if available, standard lookup otherwise
-        converted_entry.amount = converted_entry.amount_money.exchange_to(
+        # Use Money#exchange_to with custom rate if available, standard lookup otherwise.
+        # Mutate the entry in place rather than dup'ing — these instances are scoped to
+        # this sync-cache only and never persisted, so avoiding the dup eliminates a
+        # large amount of ActiveModel::Attribute allocations during sync.
+        # to_a materializes independent instances; no AR identity map is active during sync,
+        # so callers holding a reference to the same association will never see these mutations.
+        new_amount = e.amount_money.exchange_to(
           account.currency,
           date: e.date,
           custom_rate: custom_rate
         ).amount
 
-        converted_entry.currency = account.currency
-        converted_entry
+        e.amount = new_amount
+        e.currency = account.currency
+        e
       end
     end
 end

app/models/holding/forward_calculator.rb

@@ -64,7 +64,7 @@ def build_holdings(portfolio, date, price_source: nil)
           next
         end
 
-        Holding.new(
+        Holding::HoldingData.new(
           account_id: account.id,
           security_id: security_id,
           date: date,

app/models/holding/gapfillable.rb

@@ -20,9 +20,9 @@ def gapfill(holdings)
             previous_holding = holding
           else
             # Create a new holding based on the previous day's data
-            filled_holdings << Holding.new(
-              account: previous_holding.account,
-              security: previous_holding.security,
+            filled_holdings << Holding::HoldingData.new(
+              account_id: previous_holding.account_id,
+              security_id: previous_holding.security_id,
               date: date,
               qty: previous_holding.qty,
               price: previous_holding.price,

app/models/holding/holding_data.rb

@@ -0,0 +1,5 @@
+Holding::HoldingData = Struct.new(
+  :account_id, :security_id, :date,
+  :qty, :price, :currency, :amount, :cost_basis,
+  keyword_init: true
+)

app/models/holding/materializer.rb

@@ -1,6 +1,11 @@
 # "Materializes" holdings (similar to a DB materialized view, but done at the app level)
 # into a series of records we can easily query and join with other data.
 class Holding::Materializer
+  # Chunk upserts so the intermediate attribute-hash arrays
+  # (holdings_to_upsert_with_cost / _without_cost) don't sit in memory
+  # alongside the full @holdings collection. Reduces peak RSS during sync.
+  PERSIST_BATCH_SIZE = 2_000
+
   def initialize(account, strategy:, security_ids: nil)
     @account = account
     @strategy = strategy
@@ -28,9 +33,11 @@ def materialize_holdings
     # than the account or the reverse-calculated holdings.
     cleanup_stale_calculated_rows_on_latest_provider_snapshot
 
-    # Reload holdings association to clear any cached stale data
-    # This ensures subsequent Balance calculations see the fresh holdings
-    account.holdings.reload
+    # Clear the holdings association cache (without eagerly reloading) so subsequent
+    # Balance calculations rebuild from the freshly-persisted rows on demand. Using
+    # `reset` instead of `reload` avoids materializing the entire collection here when
+    # the next consumer (Balance::SyncCache) will iterate it anyway.
+    account.holdings.reset
 
     @holdings
   end
@@ -50,9 +57,16 @@ def persist_holdings
       # Load existing holdings to check locked status and source priority
       existing_holdings_map = load_existing_holdings_map
 
-      # Separate holdings into categories based on cost_basis reconciliation
-      holdings_to_upsert_with_cost = []
-      holdings_to_upsert_without_cost = []
+      # Separate holdings into categories based on cost_basis reconciliation.
+      # Use two buffers that flush at PERSIST_BATCH_SIZE so peak memory is bounded
+      # rather than accumulating the full upsert payload before writing.
+      holdings_buffer_to_upsert_with_cost = []
+      holdings_buffer_to_upsert_without_cost = []
+
+      flush = ->(buf) do
+        account.holdings.upsert_all(buf, unique_by: %i[account_id security_id date currency])
+        buf.clear
+      end
 
       @holdings.each do |holding|
         key = holding_key(holding)
@@ -74,19 +88,28 @@ def persist_holdings
           incoming_source: "calculated"
         )
 
-        base_attrs = holding.attributes
-          .slice("date", "currency", "qty", "price", "amount", "security_id")
-          .merge("account_id" => account.id, "updated_at" => current_time)
+        base_attrs = {
+          "date" => holding.date,
+          "currency" => holding.currency,
+          "qty" => holding.qty,
+          "price" => holding.price,
+          "amount" => holding.amount,
+          "security_id" => holding.security_id,
+          "account_id" => account.id,
+          "updated_at" => current_time
+        }
 
         if existing&.cost_basis_locked?
           # For locked holdings, preserve ALL cost_basis fields
-          holdings_to_upsert_without_cost << base_attrs
+          holdings_buffer_to_upsert_without_cost << base_attrs
+          flush.call(holdings_buffer_to_upsert_without_cost) if holdings_buffer_to_upsert_without_cost.size >= PERSIST_BATCH_SIZE
         elsif reconciled[:should_update] && reconciled[:cost_basis].present?
           # Update with new cost_basis and source
-          holdings_to_upsert_with_cost << base_attrs.merge(
+          holdings_buffer_to_upsert_with_cost << base_attrs.merge(
             "cost_basis" => reconciled[:cost_basis],
             "cost_basis_source" => reconciled[:cost_basis_source]
           )
+          flush.call(holdings_buffer_to_upsert_with_cost) if holdings_buffer_to_upsert_with_cost.size >= PERSIST_BATCH_SIZE
         else
           # No new calculated value — fall back to the most recent provider
           # cost_basis for this security on or before the holding date.
@@ -95,38 +118,29 @@ def persist_holdings
           preserve_existing = existing&.cost_basis.present? && %w[calculated manual].include?(existing_source)
 
           if preserve_existing
-            holdings_to_upsert_without_cost << base_attrs
+            holdings_buffer_to_upsert_without_cost << base_attrs
+            flush.call(holdings_buffer_to_upsert_without_cost) if holdings_buffer_to_upsert_without_cost.size >= PERSIST_BATCH_SIZE
           else
             carried = carry_forward_provider_cost_basis(holding)
 
             if carried && (existing&.cost_basis != carried || existing_source != "provider")
-              holdings_to_upsert_with_cost << base_attrs.merge(
+              holdings_buffer_to_upsert_with_cost << base_attrs.merge(
                 "cost_basis" => carried,
                 "cost_basis_source" => "provider"
               )
+              flush.call(holdings_buffer_to_upsert_with_cost) if holdings_buffer_to_upsert_with_cost.size >= PERSIST_BATCH_SIZE
             else
               # No cost_basis to set, or existing is better - don't touch cost_basis fields
-              holdings_to_upsert_without_cost << base_attrs
+              holdings_buffer_to_upsert_without_cost << base_attrs
+              flush.call(holdings_buffer_to_upsert_without_cost) if holdings_buffer_to_upsert_without_cost.size >= PERSIST_BATCH_SIZE
             end
           end
         end
       end
 
-      # Upsert with cost_basis updates
-      if holdings_to_upsert_with_cost.any?
-        account.holdings.upsert_all(
-          holdings_to_upsert_with_cost,
-          unique_by: %i[account_id security_id date currency]
-        )
-      end
-
-      # Upsert without cost_basis (preserves existing)
-      if holdings_to_upsert_without_cost.any?
-        account.holdings.upsert_all(
-          holdings_to_upsert_without_cost,
-          unique_by: %i[account_id security_id date currency]
-        )
-      end
+      # Flush remaining items in each buffer
+      flush.call(holdings_buffer_to_upsert_with_cost) unless holdings_buffer_to_upsert_with_cost.empty?
+      flush.call(holdings_buffer_to_upsert_without_cost) unless holdings_buffer_to_upsert_without_cost.empty?
     end
 
     def load_existing_holdings_map

app/models/holding/reverse_calculator.rb

@@ -66,7 +66,7 @@ def build_holdings(portfolio, date, price_source: nil)
           next
         end
 
-        Holding.new(
+        Holding::HoldingData.new(
           account_id: account.id,
           security_id: security_id,
           date: date,

app/models/market_data_importer.rb

@@ -78,14 +78,15 @@ def required_exchange_rate_pairs
         pair_dates[key] = [ pair_dates[key], date ].compact.min
       end
 
-      # 2. ACCOUNT-BASED PAIRS – use the account's oldest entry date
-      account_first_entry_dates = Entry.group(:account_id).minimum(:date)
-
+      # 2. ACCOUNT-BASED PAIRS – use the account's oldest entry date.
+      # The earliest entry date per account is resolved in SQL to avoid loading a
+      # potentially large Hash of all account IDs into Ruby memory.
       Account.joins(:family)
+             .joins("LEFT JOIN (SELECT account_id, MIN(date) AS first_entry_date FROM entries GROUP BY account_id) AS entry_mins ON entry_mins.account_id = accounts.id")
              .where.not("families.currency = accounts.currency")
-             .select("accounts.id, accounts.currency AS source, families.currency AS target")
+             .select("accounts.id, accounts.currency AS source, families.currency AS target, entry_mins.first_entry_date")
              .find_each do |account|
-        earliest_entry_date = account_first_entry_dates[account.id]
+        earliest_entry_date = account.first_entry_date
 
         chosen_date = [ earliest_entry_date, default_start_date ].compact.min