source-stripe-native: reduce memory usage for connected accounts

When a user's Stripe account has multiple thousand connected accounts,
memory usage limits how many accounts from which we can concurrently
capture (see 85aab13). This means any
memory efficiency improvements can improve connector performance and
allow us to increase the number of concurrent workers/subtasks allowed
in the `PriorityQueueConfig`.

This commit addresses some low-hanging fruit that have a noteable impact
on the connector's memory usage. Instead of each binding creating its
own `initial_state` and `all_account_ids` list, these can be created
once and passed into the resource creation functions.

Testing locally with ~7,000 connected accounts, I've observed a lower
memory usage when all bindings are backfilling (~80% down to ~65%). I
also want to see what the memory usage looks like when all bindings are
backfilled & we're rapidly cycling through accounts to catch them up
incrementally. I'm not confident my attempts to replicate that locally
adequately reflect actual production captures, so I'd like to merge
these memory improvements, observe memory usage in production,
then modify settings in `priority_capture.py` to take advantage of
the freshly gained memory efficiency.

Author: Alex-Bair

source-stripe-native/source_stripe_native/resources.py

@@ -1,3 +1,4 @@
+from copy import deepcopy
 from datetime import datetime, UTC, timedelta
 from logging import Logger
 import functools
@@ -127,8 +128,8 @@ def _reconcile_connector_state(
                 task.log.info(
                     f"Initializing new subtask state for account id {account_id}."
                 )
-                state.inc[account_id] = initial_state.inc[account_id]
-                state.backfill[account_id] = initial_state.backfill[account_id]
+                state.inc[account_id] = deepcopy(initial_state.inc[account_id])
+                state.backfill[account_id] = deepcopy(initial_state.backfill[account_id])
                 should_checkpoint = True
             elif not inc_state_exists and backfill_state_exists:
                 # Note: This case is to fix a legacy issue where the incremental state was not initialized
@@ -137,8 +138,8 @@ def _reconcile_connector_state(
                 task.log.info(
                     f"Backfilling subtask for account id {account_id} due to missing incremental state."
                 )
-                state.inc[account_id] = initial_state.inc[account_id]
-                state.backfill[account_id] = initial_state.backfill[account_id]
+                state.inc[account_id] = deepcopy(initial_state.inc[account_id])
+                state.backfill[account_id] = deepcopy(initial_state.backfill[account_id])
                 should_checkpoint = True
 
         if should_checkpoint:
@@ -179,6 +180,11 @@ async def all_resources(
             }
         )
 
+    all_account_ids = [platform_account_id, *connected_account_ids]
+    initial_state = _create_initial_state(
+        all_account_ids if connected_account_ids else platform_account_id
+    )
+
     for element in STREAMS:
         is_accessible_stream = True
         base_stream = element.get("stream")
@@ -197,6 +203,8 @@ async def all_resources(
                     config.start_date,
                     platform_account_id,
                     connected_account_ids,
+                    all_account_ids,
+                    initial_state,
                 )
                 if hasattr(base_stream, "EVENT_TYPES")
                 else no_events_object(
@@ -205,6 +213,8 @@ async def all_resources(
                     config.start_date,
                     platform_account_id,
                     connected_account_ids,
+                    all_account_ids,
+                    initial_state,
                 )
             )
             all_streams.append(resource)
@@ -229,6 +239,8 @@ async def all_resources(
                                     config.start_date,
                                     platform_account_id,
                                     connected_account_ids,
+                                    all_account_ids,
+                                    initial_state,
                                 )
                             )
                         case _ if child_stream.NAME in SPLIT_CHILD_STREAM_NAMES:
@@ -240,6 +252,8 @@ async def all_resources(
                                     config.start_date,
                                     platform_account_id,
                                     connected_account_ids,
+                                    all_account_ids,
+                                    initial_state,
                                 )
                             )
                         case _:
@@ -251,6 +265,8 @@ async def all_resources(
                                     config.start_date,
                                     platform_account_id,
                                     connected_account_ids,
+                                    all_account_ids,
+                                    initial_state,
                                 )
                             )
 
@@ -265,6 +281,8 @@ async def all_resources(
                 config.start_date,
                 platform_account_id,
                 connected_account_ids,
+                all_account_ids,
+                initial_state,
             )
             all_streams.append(resource)
 
@@ -300,14 +318,12 @@ def base_object(
     start_date: datetime,
     platform_account_id: str,
     connected_account_ids: list[str],
+    all_account_ids: list[str],
+    initial_state: ResourceState,
 ) -> Resource:
     """Base Object handles the default case from source-stripe-native
     It requires a single, parent stream with a valid Event API Type
     """
-    all_account_ids = [platform_account_id, *connected_account_ids]
-    initial_state = _create_initial_state(
-        all_account_ids if connected_account_ids else platform_account_id
-    )
 
     def open(
         binding: CaptureBinding[ResourceConfig],
@@ -392,17 +408,14 @@ def child_object(
     start_date: datetime,
     platform_account_id: str,
     connected_account_ids: list[str],
+    all_account_ids: list[str],
+    initial_state: ResourceState,
 ) -> Resource:
     """Child Object handles the default child case from source-stripe-native
     It requires both the parent and child stream, with the parent stream having
     a valid Event API Type
     """
 
-    all_account_ids = [platform_account_id, *connected_account_ids]
-    initial_state = _create_initial_state(
-        all_account_ids if connected_account_ids else platform_account_id
-    )
-
     def open(
         binding: CaptureBinding[ResourceConfig],
         binding_index: int,
@@ -493,18 +506,15 @@ def split_child_object(
     start_date: datetime,
     platform_account_id: str,
     connected_account_ids: list[str],
+    all_account_ids: list[str],
+    initial_state: ResourceState,
 ) -> Resource:
     """
     split_child_object handles the case where a stream is a child stream when backfilling
     but incrementally replicates based off events that contain the child stream resource directly
     in the API response. Meaning, the stream behaves like a non-chid stream incrementally.
     """
 
-    all_account_ids = [platform_account_id, *connected_account_ids]
-    initial_state = _create_initial_state(
-        all_account_ids if connected_account_ids else platform_account_id
-    )
-
     def open(
         binding: CaptureBinding[ResourceConfig],
         binding_index: int,
@@ -593,17 +603,14 @@ def usage_records(
     start_date: datetime,
     platform_account_id: str,
     connected_account_ids: list[str],
+    all_account_ids: list[str],
+    initial_state: ResourceState,
 ) -> Resource:
     """Usage Records handles a specific stream (UsageRecords).
     This is required since Usage Records is a child stream from SubscriptionItem
     and requires special processing.
     """
 
-    all_account_ids = [platform_account_id, *connected_account_ids]
-    initial_state = _create_initial_state(
-        all_account_ids if connected_account_ids else platform_account_id
-    )
-
     def open(
         binding: CaptureBinding[ResourceConfig],
         binding_index: int,
@@ -692,18 +699,15 @@ def no_events_object(
     start_date: datetime,
     platform_account_id: str,
     connected_account_ids: list[str],
+    all_account_ids: list[str],
+    initial_state: ResourceState,
 ) -> Resource:
     """No Events Object handles a edge-case from source-stripe-native,
     where the given parent stream does not contain a valid Events API type.
     It requires a single, parent stream with a valid list all API endpoint.
     It works very similar to the base object, but without the use of the Events APi.
     """
 
-    all_account_ids = [platform_account_id, *connected_account_ids]
-    initial_state = _create_initial_state(
-        all_account_ids if connected_account_ids else platform_account_id
-    )
-
     def open(
         binding: CaptureBinding[ResourceConfig],
         binding_index: int,