mirror of
https://github.com/we-promise/sure.git
synced 2026-05-28 15:04:57 +00:00
fix(jobs): delegate recurring-transaction sync gate to Sync.for_family (#1975)
* fix(jobs): delegate recurring-transaction sync gate to Sync.for_family `IdentifyRecurringTransactionsJob#family_has_incomplete_syncs?` hand-rolled the list of provider `*_items` associations it polled — plaid, simplefin, lunchflow, enable_banking, sophtron — missing nine other `Syncable` provider concerns on `Family`: coinbase, binance, kraken, coinstats, snaptrade, mercury, brex, indexa_capital, ibkr. When a sync on any of those nine was in flight, the debounce gate fell through and `RecurringTransaction::Identifier` ran against a partial dataset; the follow-up re-enqueue then hit the `find_or_initialize_by` upsert path and inherited the stale `occurrence_count`. Same drift pattern that bolted sophtron on as the 5th entry (#591) was already an iteration of. The maintainers' own `Sync.for_family` (sync.rb:61) already enumerates every `*_items` association via `Family.reflect_on_all_associations(:has_many)` filtered by inclusion of `Syncable` — exactly the helper the gate should delegate to so the list cannot drift again. - Add `Sync.any_incomplete_for?(family)` class method that wraps `for_family(family).incomplete.exists?`. - Rewrite `family_has_incomplete_syncs?` to delegate. 14 lines → 1. - New test file `test/jobs/identify_recurring_transactions_job_test.rb` covers in-flight Coinbase + Mercury (gate fires), idle (identifier runs), missing family, and superseded-by-newer-schedule. - `test/models/sync_test.rb` gets 2 new tests pinning `any_incomplete_for?` against a provider `_items` sync and a family-itself sync. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> * test(jobs): stub Rails.cache.read for supersession test (NullStore in test env) `Rails.cache` is `ActiveSupport::Cache::NullStore` in the Rails test env, so the previous test's `Rails.cache.write(cache_key, @scheduled_at + 10, ...)` was a no-op and `Rails.cache.read(cache_key)` returned `nil`. The supersession short-circuit `return if latest_scheduled && latest_scheduled > scheduled_at` then fell through, the job proceeded to invoke `RecurringTransaction::Identifier`, and the Mocha `.expects(:identify_recurring_patterns).never` failed in CI. Switch to `Rails.cache.stubs(:read).with(cache_key).returns(...)` — the same idiom `test/models/provider/twelve_data_test.rb:186-197` already uses for the cache layer. Add an `assert_nil` on the bare `perform` return so Minitest's assertion counter sees an explicit assertion (silences the "missing assertions" warning). No production-code change. Behavior under test is unchanged; only the test mechanism for simulating "newer scheduled run already in cache" is fixed. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -42,21 +42,13 @@ class IdentifyRecurringTransactionsJob < ApplicationJob
|
||||
"recurring_transaction_identify:#{family_id}"
|
||||
end
|
||||
|
||||
# Debounce gate: delegate to `Sync.any_incomplete_for?`, which polls every
|
||||
# `Syncable` provider association on `Family` via reflection. The previous
|
||||
# hand-rolled list covered only 5 of the 14 `*_items` associations on
|
||||
# `Family`, so a Coinbase/Mercury/Brex/etc. sync in flight silently
|
||||
# bypassed this gate and let the identifier run against a partial dataset.
|
||||
def family_has_incomplete_syncs?(family)
|
||||
# Check family's own syncs
|
||||
return true if family.syncs.incomplete.exists?
|
||||
|
||||
# Check all provider items' syncs
|
||||
return true if family.plaid_items.joins(:syncs).merge(Sync.incomplete).exists? if family.respond_to?(:plaid_items)
|
||||
return true if family.simplefin_items.joins(:syncs).merge(Sync.incomplete).exists? if family.respond_to?(:simplefin_items)
|
||||
return true if family.lunchflow_items.joins(:syncs).merge(Sync.incomplete).exists? if family.respond_to?(:lunchflow_items)
|
||||
return true if family.enable_banking_items.joins(:syncs).merge(Sync.incomplete).exists? if family.respond_to?(:enable_banking_items)
|
||||
return true if family.sophtron_items.joins(:syncs).merge(Sync.incomplete).exists? if family.respond_to?(:sophtron_items)
|
||||
|
||||
# Check accounts' syncs
|
||||
return true if family.accounts.joins(:syncs).merge(Sync.incomplete).exists?
|
||||
|
||||
false
|
||||
Sync.any_incomplete_for?(family)
|
||||
end
|
||||
|
||||
def with_advisory_lock(family_id)
|
||||
|
||||
@@ -71,6 +71,14 @@ class Sync < ApplicationRecord
|
||||
query
|
||||
end
|
||||
|
||||
# True iff the family has any pending/syncing Sync — across its own row,
|
||||
# its accounts, and every Syncable provider `*_items` association. Built
|
||||
# on `for_family` so new provider integrations are picked up automatically
|
||||
# via `family_syncable_associations` reflection (no hand-rolled list).
|
||||
def any_incomplete_for?(family)
|
||||
for_family(family).incomplete.exists?
|
||||
end
|
||||
|
||||
private
|
||||
def account_syncable_ids(family, resource_owner)
|
||||
(resource_owner ? resource_owner.accessible_accounts : family.accounts)
|
||||
|
||||
57
test/jobs/identify_recurring_transactions_job_test.rb
Normal file
57
test/jobs/identify_recurring_transactions_job_test.rb
Normal file
@@ -0,0 +1,57 @@
|
||||
require "test_helper"
|
||||
|
||||
class IdentifyRecurringTransactionsJobTest < ActiveJob::TestCase
|
||||
setup do
|
||||
@family = families(:dylan_family)
|
||||
@scheduled_at = Time.current.to_f
|
||||
end
|
||||
|
||||
test "skips identification while a Coinbase provider sync is in flight" do
|
||||
coinbase_item = @family.coinbase_items.create!(
|
||||
name: "Coinbase Pro",
|
||||
api_key: "test-api-key-#{SecureRandom.hex(4)}",
|
||||
api_secret: "test-api-secret-#{SecureRandom.hex(8)}"
|
||||
)
|
||||
Sync.create!(syncable: coinbase_item, status: :syncing)
|
||||
|
||||
RecurringTransaction::Identifier.any_instance.expects(:identify_recurring_patterns).never
|
||||
|
||||
IdentifyRecurringTransactionsJob.new.perform(@family.id, @scheduled_at)
|
||||
end
|
||||
|
||||
test "skips identification while a Mercury provider sync is in flight" do
|
||||
mercury_item = mercury_items(:one)
|
||||
Sync.create!(syncable: mercury_item, status: :pending)
|
||||
|
||||
RecurringTransaction::Identifier.any_instance.expects(:identify_recurring_patterns).never
|
||||
|
||||
IdentifyRecurringTransactionsJob.new.perform(@family.id, @scheduled_at)
|
||||
end
|
||||
|
||||
test "runs identification when no provider syncs are in flight" do
|
||||
# Sanity: there are no incomplete syncs in the fixture set by default.
|
||||
Sync.for_family(@family).incomplete.find_each(&:destroy)
|
||||
|
||||
RecurringTransaction::Identifier.any_instance.expects(:identify_recurring_patterns).once
|
||||
|
||||
IdentifyRecurringTransactionsJob.new.perform(@family.id, @scheduled_at)
|
||||
end
|
||||
|
||||
test "skips when family is missing" do
|
||||
RecurringTransaction::Identifier.any_instance.expects(:identify_recurring_patterns).never
|
||||
|
||||
IdentifyRecurringTransactionsJob.new.perform(SecureRandom.uuid, @scheduled_at)
|
||||
end
|
||||
|
||||
test "skips when a newer scheduled run supersedes this one" do
|
||||
# Rails.cache is NullStore in the test env (writes are no-ops), so we stub
|
||||
# the read directly to simulate a newer scheduled-at landing in the cache
|
||||
# between this job being enqueued and being picked up.
|
||||
cache_key = "recurring_transaction_identify:#{@family.id}"
|
||||
Rails.cache.stubs(:read).with(cache_key).returns(@scheduled_at + 10)
|
||||
|
||||
RecurringTransaction::Identifier.any_instance.expects(:identify_recurring_patterns).never
|
||||
|
||||
assert_nil IdentifyRecurringTransactionsJob.new.perform(@family.id, @scheduled_at)
|
||||
end
|
||||
end
|
||||
@@ -234,6 +234,29 @@ class SyncTest < ActiveSupport::TestCase
|
||||
assert_equal syncs.map(&:id).sort, Sync.for_family(family).where(id: syncs.map(&:id)).pluck(:id).sort
|
||||
end
|
||||
|
||||
test "any_incomplete_for? fires on a Sync against any Syncable provider item association" do
|
||||
family = families(:dylan_family)
|
||||
Sync.for_family(family).incomplete.find_each(&:destroy)
|
||||
assert_not Sync.any_incomplete_for?(family)
|
||||
|
||||
mercury_item = mercury_items(:one)
|
||||
incomplete = Sync.create!(syncable: mercury_item, status: :pending)
|
||||
assert Sync.any_incomplete_for?(family),
|
||||
"any_incomplete_for? should report true for an in-flight Mercury sync"
|
||||
|
||||
incomplete.update!(status: :completed)
|
||||
assert_not Sync.any_incomplete_for?(family)
|
||||
end
|
||||
|
||||
test "any_incomplete_for? fires on a Sync against the family itself" do
|
||||
family = families(:dylan_family)
|
||||
Sync.for_family(family).incomplete.find_each(&:destroy)
|
||||
assert_not Sync.any_incomplete_for?(family)
|
||||
|
||||
Sync.create!(syncable: family, status: :syncing)
|
||||
assert Sync.any_incomplete_for?(family)
|
||||
end
|
||||
|
||||
test "api error payload is present for failed syncs without raw error text" do
|
||||
sync = Sync.create!(syncable: accounts(:depository), status: :failed)
|
||||
|
||||
|
||||
Reference in New Issue
Block a user