From 5a798435b3b816df623f0e42f6cd0d1bfc81449d Mon Sep 17 00:00:00 2001 From: Tim Katz <40766869+timkatz@users.noreply.github.com> Date: Thu, 27 Aug 2026 12:34:09 -0700 Subject: [PATCH] Fix user-triggered Plaid transaction refresh (#3206) * Fix user-triggered Plaid transaction refresh Request a fresh Plaid institution update for explicit user syncs, then poll the saved cursor with bounded retries so private self-hosted instances do not depend on webhooks. Preserve the existing immediate sync and coalesce repeated refresh requests.\n\nCloses #3204 * Address Plaid refresh concurrency races * Preserve Plaid refresh handoff on retry exhaustion * Release Plaid refresh lease on enqueue errors Ensure adapter exceptions cannot leave the shared refresh cooldown occupied when no job was queued. --- app/controllers/accounts_controller.rb | 9 ++- app/controllers/plaid_items_controller.rb | 2 +- ...transactions_refresh_follow_up_sync_job.rb | 29 ++++++++ app/jobs/plaid_transactions_refresh_job.rb | 25 +++++++ .../plaid_transactions_refresh_poll_job.rb | 47 ++++++++++++ app/models/plaid_item.rb | 43 +++++++++++ app/models/provider/plaid.rb | 5 ++ test/controllers/accounts_controller_test.rb | 11 ++- .../plaid_items_controller_test.rb | 2 +- ...actions_refresh_follow_up_sync_job_test.rb | 39 ++++++++++ .../plaid_transactions_refresh_job_test.rb | 33 +++++++++ ...laid_transactions_refresh_poll_job_test.rb | 72 +++++++++++++++++++ test/models/plaid_item_test.rb | 68 ++++++++++++++++++ test/models/provider/plaid_test.rb | 8 +++ 14 files changed, 389 insertions(+), 4 deletions(-) create mode 100644 app/jobs/plaid_transactions_refresh_follow_up_sync_job.rb create mode 100644 app/jobs/plaid_transactions_refresh_job.rb create mode 100644 app/jobs/plaid_transactions_refresh_poll_job.rb create mode 100644 test/jobs/plaid_transactions_refresh_follow_up_sync_job_test.rb create mode 100644 test/jobs/plaid_transactions_refresh_job_test.rb create mode 100644 test/jobs/plaid_transactions_refresh_poll_job_test.rb diff --git a/app/controllers/accounts_controller.rb b/app/controllers/accounts_controller.rb index 727af4dce..ad3395819 100644 --- a/app/controllers/accounts_controller.rb +++ b/app/controllers/accounts_controller.rb @@ -63,6 +63,7 @@ class AccountsController < ApplicationController end def sync_all + family.plaid_items.syncable.each(&:request_transactions_refresh_later) family.sync_later redirect_to accounts_path, notice: t("accounts.sync_all.syncing") end @@ -135,7 +136,13 @@ class AccountsController < ApplicationController # Each provider item will trigger an account sync when complete @account.account_providers.each do |account_provider| item = account_provider.adapter&.item - item&.sync_later if item && !item.syncing? + next unless item && !item.syncing? + + if item.is_a?(PlaidItem) + item.sync_later_with_provider_refresh + else + item.sync_later + end end else # Manual accounts just need balance materialization diff --git a/app/controllers/plaid_items_controller.rb b/app/controllers/plaid_items_controller.rb index 47d38282f..bd5844568 100644 --- a/app/controllers/plaid_items_controller.rb +++ b/app/controllers/plaid_items_controller.rb @@ -46,7 +46,7 @@ class PlaidItemsController < ApplicationController end def sync - @plaid_item.sync_later_with_follow_up + @plaid_item.sync_later_with_provider_refresh respond_to do |format| format.html { redirect_back_or_to accounts_path } diff --git a/app/jobs/plaid_transactions_refresh_follow_up_sync_job.rb b/app/jobs/plaid_transactions_refresh_follow_up_sync_job.rb new file mode 100644 index 000000000..a26178392 --- /dev/null +++ b/app/jobs/plaid_transactions_refresh_follow_up_sync_job.rb @@ -0,0 +1,29 @@ +class PlaidTransactionsRefreshFollowUpSyncJob < ApplicationJob + queue_as :high_priority + + RETRY_DELAY = 10.seconds + MAX_ATTEMPTS = 36 + + def perform(plaid_item, attempts_remaining: MAX_ATTEMPTS) + if plaid_item.syncs.visible.exists? + if attempts_remaining.positive? + self.class.set(wait: RETRY_DELAY).perform_later(plaid_item, attempts_remaining: attempts_remaining - 1) + else + DebugLogEntry.capture( + category: "provider_sync", + level: "warn", + message: "Plaid transaction refresh follow-up exhausted its wait for an active sync", + source: self.class.name, + provider_key: "plaid", + family: plaid_item.family + ) + + plaid_item.sync_later + end + + return + end + + plaid_item.sync_later + end +end diff --git a/app/jobs/plaid_transactions_refresh_job.rb b/app/jobs/plaid_transactions_refresh_job.rb new file mode 100644 index 000000000..b6571ed53 --- /dev/null +++ b/app/jobs/plaid_transactions_refresh_job.rb @@ -0,0 +1,25 @@ +class PlaidTransactionsRefreshJob < ApplicationJob + queue_as :high_priority + + def perform(plaid_item) + cursor = plaid_item.next_cursor + + begin + plaid_item.plaid_provider.refresh_transactions(plaid_item.access_token) + rescue => error + DebugLogEntry.capture( + category: "provider_sync", + level: "warn", + message: "Plaid transaction refresh request did not return successfully; checking for asynchronous completion", + source: self.class.name, + provider_key: "plaid", + family: plaid_item.family, + metadata: { error_class: error.class.name } + ) + end + + PlaidTransactionsRefreshPollJob + .set(wait: PlaidTransactionsRefreshPollJob::POLL_INTERVAL) + .perform_later(plaid_item, cursor: cursor) + end +end diff --git a/app/jobs/plaid_transactions_refresh_poll_job.rb b/app/jobs/plaid_transactions_refresh_poll_job.rb new file mode 100644 index 000000000..77deb14bd --- /dev/null +++ b/app/jobs/plaid_transactions_refresh_poll_job.rb @@ -0,0 +1,47 @@ +class PlaidTransactionsRefreshPollJob < ApplicationJob + queue_as :high_priority + + POLL_INTERVAL = 30.seconds + MAX_ATTEMPTS = 6 + + def perform(plaid_item, cursor:, attempts_remaining: MAX_ATTEMPTS) + plaid_item.reload + cursor = plaid_item.next_cursor if plaid_item.next_cursor != cursor + + transactions = plaid_item.plaid_provider.get_transactions( + plaid_item.access_token, + next_cursor: cursor + ) + + if transactions.cursor != cursor + PlaidTransactionsRefreshFollowUpSyncJob.perform_later(plaid_item) + elsif attempts_remaining.to_i > 1 + self.class + .set(wait: POLL_INTERVAL) + .perform_later(plaid_item, cursor: cursor, attempts_remaining: attempts_remaining.to_i - 1) + else + DebugLogEntry.capture( + category: "provider_sync", + level: "warn", + message: "Plaid transaction refresh completed without advancing the transaction cursor", + source: self.class.name, + provider_key: "plaid", + family: plaid_item.family + ) + + PlaidTransactionsRefreshFollowUpSyncJob.perform_later(plaid_item) + end + rescue => error + DebugLogEntry.capture( + category: "provider_sync", + level: "warn", + message: "Plaid transaction refresh polling failed; falling back to a normal sync", + source: self.class.name, + provider_key: "plaid", + family: plaid_item.family, + metadata: { error_class: error.class.name } + ) + + PlaidTransactionsRefreshFollowUpSyncJob.perform_later(plaid_item) + end +end diff --git a/app/models/plaid_item.rb b/app/models/plaid_item.rb index f8f3ca16d..68a3a16f1 100644 --- a/app/models/plaid_item.rb +++ b/app/models/plaid_item.rb @@ -27,6 +27,8 @@ class PlaidItem < ApplicationRecord scope :ordered, -> { order(created_at: :desc) } scope :needs_update, -> { where(status: :requires_update) } + TRANSACTIONS_REFRESH_COOLDOWN = 5.minutes + # Get accounts from both new and legacy systems def accounts @accounts ||= plaid_accounts @@ -81,6 +83,34 @@ class PlaidItem < ApplicationRecord DestroyJob.perform_later(self) end + def request_transactions_refresh_later + return unless supports_product?("transactions") + return unless shared_transactions_refresh_cache? + + refresh_requested = Rails.cache.write( + transactions_refresh_cache_key, + true, + expires_in: TRANSACTIONS_REFRESH_COOLDOWN, + unless_exist: true + ) + + return unless refresh_requested + + enqueued_job = begin + PlaidTransactionsRefreshJob.perform_later(self) + rescue + Rails.cache.delete(transactions_refresh_cache_key) + raise + end + + Rails.cache.delete(transactions_refresh_cache_key) unless enqueued_job + end + + def sync_later_with_provider_refresh + request_transactions_refresh_later + sync_later_with_follow_up + end + def import_latest_plaid_data PlaidItem::Importer.new(self, plaid_provider: plaid_provider).import end @@ -133,6 +163,19 @@ class PlaidItem < ApplicationRecord end private + def transactions_refresh_cache_key + "plaid_item:#{id}:transactions_refresh_requested" + end + + def shared_transactions_refresh_cache? + shared_cache = Rails.cache.is_a?(ActiveSupport::Cache::RedisCacheStore) || + Rails.cache.is_a?(ActiveSupport::Cache::MemCacheStore) || + Rails.cache.class.name == "SolidCache::Store" + + Rails.logger.warn("Plaid transaction refresh requires a shared Rails cache store") unless shared_cache + shared_cache + end + def remove_plaid_item return unless plaid_provider.present? diff --git a/app/models/provider/plaid.rb b/app/models/provider/plaid.rb index 1e341a26e..a9a13fef9 100644 --- a/app/models/provider/plaid.rb +++ b/app/models/provider/plaid.rb @@ -117,6 +117,11 @@ class Provider::Plaid TransactionSyncResponse.new(added:, modified:, removed:, cursor:) end + def refresh_transactions(access_token) + request = Plaid::TransactionsRefreshRequest.new(access_token: access_token) + client.transactions_refresh(request) + end + def get_item_investments(access_token, start_date: nil, end_date: Date.current) start_date = start_date || MAX_HISTORY_DAYS.days.ago.to_date holdings, holding_securities = get_item_holdings(access_token: access_token) diff --git a/test/controllers/accounts_controller_test.rb b/test/controllers/accounts_controller_test.rb index 7b35a3968..9ae0139a4 100644 --- a/test/controllers/accounts_controller_test.rb +++ b/test/controllers/accounts_controller_test.rb @@ -79,6 +79,15 @@ class AccountsControllerTest < ActionDispatch::IntegrationTest assert_response :success end + test "sync all requests fresh Plaid transactions before syncing the family" do + PlaidItem.any_instance.expects(:request_transactions_refresh_later).once + Family.any_instance.expects(:sync_later).once + + post sync_all_accounts_url + + assert_redirected_to accounts_url + end + test "show avoids N+1 transfer queries across paginated entries" do queries = capture_sql_queries { get account_url(@account) } assert_response :success @@ -353,7 +362,7 @@ class AccountsControllerTest < ActionDispatch::IntegrationTest # Mock at the class level since controller loads account from DB Account.any_instance.expects(:syncing?).returns(false) PlaidItem.any_instance.expects(:syncing?).returns(false) - PlaidItem.any_instance.expects(:sync_later).once + PlaidItem.any_instance.expects(:sync_later_with_provider_refresh).once post sync_account_url(@account) assert_redirected_to account_url(@account) diff --git a/test/controllers/plaid_items_controller_test.rb b/test/controllers/plaid_items_controller_test.rb index 8f34b653a..8084038c7 100644 --- a/test/controllers/plaid_items_controller_test.rb +++ b/test/controllers/plaid_items_controller_test.rb @@ -119,7 +119,7 @@ class PlaidItemsControllerTest < ActionDispatch::IntegrationTest test "sync" do plaid_item = plaid_items(:one) - PlaidItem.any_instance.expects(:sync_later_with_follow_up).once + PlaidItem.any_instance.expects(:sync_later_with_provider_refresh).once post sync_plaid_item_url(plaid_item) diff --git a/test/jobs/plaid_transactions_refresh_follow_up_sync_job_test.rb b/test/jobs/plaid_transactions_refresh_follow_up_sync_job_test.rb new file mode 100644 index 000000000..0f827fc24 --- /dev/null +++ b/test/jobs/plaid_transactions_refresh_follow_up_sync_job_test.rb @@ -0,0 +1,39 @@ +require "test_helper" + +class PlaidTransactionsRefreshFollowUpSyncJobTest < ActiveJob::TestCase + test "queues a distinct sync after the active sync has finished" do + item = plaid_items(:one) + + assert_difference "item.syncs.count", 1 do + PlaidTransactionsRefreshFollowUpSyncJob.perform_now(item) + end + end + + test "retries while an item sync is in progress" do + item = plaid_items(:one) + active_sync = item.syncs.create! + active_sync.start! + + assert_no_difference "item.syncs.count" do + assert_enqueued_with job: PlaidTransactionsRefreshFollowUpSyncJob do + PlaidTransactionsRefreshFollowUpSyncJob.perform_now(item) + end + end + end + + test "records exhausted retries and still hands off to sync_later" do + item = plaid_items(:one) + active_sync = item.syncs.create! + active_sync.start! + item.expects(:sync_later).once + + assert_difference "DebugLogEntry.count", 1 do + PlaidTransactionsRefreshFollowUpSyncJob.perform_now(item, attempts_remaining: 0) + end + + entry = DebugLogEntry.order(:created_at).last + assert_equal "provider_sync", entry.category + assert_equal "PlaidTransactionsRefreshFollowUpSyncJob", entry.source + assert_equal item.family, entry.family + end +end diff --git a/test/jobs/plaid_transactions_refresh_job_test.rb b/test/jobs/plaid_transactions_refresh_job_test.rb new file mode 100644 index 000000000..43ad39320 --- /dev/null +++ b/test/jobs/plaid_transactions_refresh_job_test.rb @@ -0,0 +1,33 @@ +require "test_helper" + +class PlaidTransactionsRefreshJobTest < ActiveJob::TestCase + setup do + @plaid_item = plaid_items(:one) + @plaid_item.update!(next_cursor: "saved-cursor") + @provider = mock + PlaidItem.any_instance.stubs(:plaid_provider).returns(@provider) + end + + test "requests refresh and schedules cursor polling" do + @provider.expects(:refresh_transactions).with(@plaid_item.access_token) + + assert_enqueued_with( + job: PlaidTransactionsRefreshPollJob, + args: [ @plaid_item, { cursor: "saved-cursor" } ] + ) do + PlaidTransactionsRefreshJob.perform_now(@plaid_item) + end + end + + test "still polls after an ambiguous refresh request failure" do + @provider.expects(:refresh_transactions).raises(Timeout::Error) + + assert_difference "DebugLogEntry.count", 1 do + assert_enqueued_with(job: PlaidTransactionsRefreshPollJob) do + PlaidTransactionsRefreshJob.perform_now(@plaid_item) + end + end + + assert_equal "Timeout::Error", DebugLogEntry.order(:created_at).last.metadata["error_class"] + end +end diff --git a/test/jobs/plaid_transactions_refresh_poll_job_test.rb b/test/jobs/plaid_transactions_refresh_poll_job_test.rb new file mode 100644 index 000000000..27a8713d1 --- /dev/null +++ b/test/jobs/plaid_transactions_refresh_poll_job_test.rb @@ -0,0 +1,72 @@ +require "test_helper" + +class PlaidTransactionsRefreshPollJobTest < ActiveJob::TestCase + setup do + @plaid_item = plaid_items(:one) + @plaid_item.update!(next_cursor: "saved-cursor") + @provider = mock + PlaidItem.any_instance.stubs(:plaid_provider).returns(@provider) + end + + test "continues polling from a cursor advanced by a concurrent sync" do + @plaid_item.update!(next_cursor: "advanced-cursor") + @provider.expects(:get_transactions) + .with(@plaid_item.access_token, next_cursor: "advanced-cursor") + .returns(stub(cursor: "refreshed-cursor")) + + assert_enqueued_with(job: SyncJob) do + perform_enqueued_jobs(only: PlaidTransactionsRefreshFollowUpSyncJob) do + PlaidTransactionsRefreshPollJob.perform_now(@plaid_item, cursor: "saved-cursor") + end + end + end + + test "schedules normal sync when refreshed deltas are available" do + @provider.expects(:get_transactions) + .with(@plaid_item.access_token, next_cursor: "saved-cursor") + .returns(stub(cursor: "advanced-cursor")) + + assert_enqueued_with(job: PlaidTransactionsRefreshFollowUpSyncJob) do + PlaidTransactionsRefreshPollJob.perform_now(@plaid_item, cursor: "saved-cursor") + end + end + + test "polls again while the cursor has not advanced" do + @provider.expects(:get_transactions).returns(stub(cursor: "saved-cursor")) + + assert_enqueued_with( + job: PlaidTransactionsRefreshPollJob, + args: [ @plaid_item, { cursor: "saved-cursor", attempts_remaining: 1 } ] + ) do + PlaidTransactionsRefreshPollJob.perform_now( + @plaid_item, + cursor: "saved-cursor", + attempts_remaining: 2 + ) + end + end + + test "falls back to normal sync when polling is exhausted" do + @provider.expects(:get_transactions).returns(stub(cursor: "saved-cursor")) + + assert_difference "DebugLogEntry.count", 1 do + assert_enqueued_with(job: PlaidTransactionsRefreshFollowUpSyncJob) do + PlaidTransactionsRefreshPollJob.perform_now( + @plaid_item, + cursor: "saved-cursor", + attempts_remaining: 1 + ) + end + end + end + + test "falls back to normal sync when polling fails" do + @provider.expects(:get_transactions).raises(Plaid::ApiError.new(code: 500)) + + assert_difference "DebugLogEntry.count", 1 do + assert_enqueued_with(job: PlaidTransactionsRefreshFollowUpSyncJob) do + PlaidTransactionsRefreshPollJob.perform_now(@plaid_item, cursor: "saved-cursor") + end + end + end +end diff --git a/test/models/plaid_item_test.rb b/test/models/plaid_item_test.rb index 117ec0ad9..969bd1c8a 100644 --- a/test/models/plaid_item_test.rb +++ b/test/models/plaid_item_test.rb @@ -113,4 +113,72 @@ class PlaidItemTest < ActiveSupport::TestCase end assert_predicate @plaid_item.reload, :good? end + + test "user sync requests a provider refresh when cooldown lease is acquired" do + @plaid_item.stubs(:shared_transactions_refresh_cache?).returns(true) + Rails.cache.expects(:write).with( + "plaid_item:#{@plaid_item.id}:transactions_refresh_requested", + true, + expires_in: PlaidItem::TRANSACTIONS_REFRESH_COOLDOWN, + unless_exist: true + ).returns(true) + + assert_enqueued_with(job: PlaidTransactionsRefreshJob, args: [ @plaid_item ]) do + @plaid_item.request_transactions_refresh_later + end + end + + test "user sync does not duplicate a recent provider refresh request" do + @plaid_item.stubs(:shared_transactions_refresh_cache?).returns(true) + Rails.cache.stubs(:write).returns(false) + + assert_no_enqueued_jobs only: PlaidTransactionsRefreshJob do + @plaid_item.request_transactions_refresh_later + end + end + + test "user sync does not request refresh without the transactions product" do + @plaid_item.update!(billed_products: [ "investments" ]) + + Rails.cache.expects(:write).never + assert_no_enqueued_jobs only: PlaidTransactionsRefreshJob do + @plaid_item.request_transactions_refresh_later + end + end + + test "user sync rejects provider refresh without a shared cache" do + @plaid_item.stubs(:shared_transactions_refresh_cache?).returns(false) + + Rails.cache.expects(:write).never + assert_no_enqueued_jobs only: PlaidTransactionsRefreshJob do + @plaid_item.request_transactions_refresh_later + end + end + + test "user sync releases cooldown lease when refresh job is not enqueued" do + @plaid_item.stubs(:shared_transactions_refresh_cache?).returns(true) + Rails.cache.stubs(:write).returns(true) + PlaidTransactionsRefreshJob.stubs(:perform_later).returns(false) + Rails.cache.expects(:delete).with("plaid_item:#{@plaid_item.id}:transactions_refresh_requested") + + @plaid_item.request_transactions_refresh_later + end + + test "user sync releases cooldown lease when refresh job enqueue raises" do + @plaid_item.stubs(:shared_transactions_refresh_cache?).returns(true) + Rails.cache.stubs(:write).returns(true) + PlaidTransactionsRefreshJob.stubs(:perform_later).raises(RedisClient::Error, "Redis unavailable") + Rails.cache.expects(:delete).with("plaid_item:#{@plaid_item.id}:transactions_refresh_requested") + + assert_raises RedisClient::Error do + @plaid_item.request_transactions_refresh_later + end + end + + test "user sync preserves follow-up sync while requesting provider refresh" do + @plaid_item.expects(:request_transactions_refresh_later).once + @plaid_item.expects(:sync_later_with_follow_up).once + + @plaid_item.sync_later_with_provider_refresh + end end diff --git a/test/models/provider/plaid_test.rb b/test/models/provider/plaid_test.rb index 83ecd6e9b..7e71c6d4c 100644 --- a/test/models/provider/plaid_test.rb +++ b/test/models/provider/plaid_test.rb @@ -81,6 +81,14 @@ class Provider::PlaidTest < ActiveSupport::TestCase end end + test "requests a transaction refresh" do + @plaid.client.expects(:transactions_refresh).with do |request| + request.access_token == "access-token" + end + + @plaid.refresh_transactions("access-token") + end + test "gets item investments" do VCR.use_cassette("plaid/get_item_investments") do access_token = get_access_token