32427 Update reserve corps logic to resolve db lock issue and other perf tweaks for migrating batches larger than 10K with migration filter on

[argush3]

Issue #: /bcgov/entity#32427

Description of changes:

Fixes tombstone flow getting “stuck reserving 1000 corps” by making reservation/count logic match the actual filtered cohort and by optimizing the reservable selection query for end-of-batch performance.

Key Changes

• Accurate reservable count: New count query mirrors get_unprocessed_corps_query filters (mig batch/group, corp types, active state, and “not already reserved/processed”). Flow now uses this count and exits early when 0.
• Faster reservable selection: When USE_MIGRATION_FILTER=True, selection drives from mig_corp_batch and uses a proper anti-join (cp.corp_num IS NULL / NOT EXISTS semantics). Selects only needed columns.
• Skip account_ids work unless needed: Only compute/persist account_ids during reservation when AFFILIATE_ENTITY (or fallback CSV) is enabled.
• DB indexes: Adds missing indexes on mig_corp_batch(mig_batch_id) and (mig_batch_id, corp_num) (plus mig_batch(mig_group_id)).
• Safety/observability: Adds reservation timing logs and optional RESERVE_STATEMENT_TIMEOUT_MS to avoid indefinite waits.
• Misc cleanup

Impact

Reservation time remains stable as the cohort drains, scheduled runs stop hammering reservation when exhausted, and affiliate-related aggregation work is avoided when not required.

Test Notes

Able to run migrate 10k businesses with following setup locally in around 11 minutes

RESERVE_STATEMENT_TIMEOUT_MS = 20000

TOMBSTONE_BATCHES=40
TOMBSTONE_BATCH_SIZE=25

USE_MIGRATION_FILTER=True
MIG_GROUP_IDS=34
MIG_BATCH_IDS=112

prefect work-pool create "tombstone-pool" --type process && \
prefect work-pool set-concurrency-limit "tombstone-pool" 10 && \
prefect worker start --pool "tombstone-pool" --type process

@flow(
    name='Corps-Tombstone-Migrate-Flow',
    log_prints=True,
    persist_result=False,
    # Parallel Migration - Setup
    task_runner=ConcurrentTaskRunner(max_workers=100)
)
def tombstone_flow():

   flow_source = Path(__file__).parent

    # 3. OTHERs deployment setup
    # subquery = subqueries[3]
    tombstone_flow.from_source(
        source=flow_source,
        entrypoint="corps_tombstone_flow.py:tombstone_flow"
    ).deploy(
        name="tombstone-deployment",
        tags=["tombstone-migration"],
        work_pool_name="tombstone-pool",
        interval=timedelta(seconds=40)  # Run every x seconds
    )

By submitting this pull request, I confirm that you can use, modify, copy, and redistribute this contribution, under the terms of the lear license (Apache 2.0).