source-salesforce-native: backfill formula fields on a schedule

Salesforce formula fields are calculated at query time - they are not
saved in Salesforce's database, so they don't have any prior state. This
means that any updates to formula fields don't always cause an object's
cursor field to update, and formula fields can be outdated. This is known
to happen when formula fields are calculated based on data external to
its object (ex: other objects, a global variable, specific times, etc.).

To fix this, we leverage the recently added `RecurringFetchPageFn` type
of `fetch_page` & the `ResourceConfigWithSchedule` to backfill formula
fields at some cadence. A top level merge reduction strategy is used for
these collections in order to merge in the updated formula fields. The
default schedule is currently once a day at 23:55 UTC, but that can be
changed as we learn more about how frequently users want these scheduled
formula field backfills to occur & what the typical API limit impact
looks like.

Author: Alex-Bair

source-salesforce-native/config.yaml

@@ -3,6 +3,7 @@ credentials:
     client_id_sops: ENC[AES256_GCM,data:fMXpKq4TrDG9vbkRFa7EYhOsdk6FtTLLT+jU5hokrHEB2FF8h39CquHiOp0+El0Mg/4zQMTdJ5wDp90WKSPa9yxzaMAi97ExkqfNWPM2FNAK2KmAYg==,iv:8lNVaJITypPnTuUZM4kMXUMdJglGFdkDYGNGhyPsxVU=,tag:yT1ENGiQzGZKM8DCT8gS0g==,type:str]
     client_secret_sops: ENC[AES256_GCM,data:jJgwo0E280bY9JApzBDCXuFaEbnnMMhtT0zYrwybJcOtZc+6RXNkxtQSbXvw9DpAAC6jbGEXMMc6yONx4y21ww==,iv:KqFuPgJzjVgDdOdKGRrnvEda/ZUUBvvlBjUcEdiALeg=,tag:M2RoEtnWixP86GOv/uwMIw==,type:str]
     refresh_token_sops: ENC[AES256_GCM,data:vqvATUauWsPBTFv1wTO7DljL4zOADONNj7+euRIQkXlB6nQgFIX11uH6tFrhhThAT99x5ipqRSxzzRgbGb/bd++AybniGSIy/Cw6AvPUDvDYAJ8yRX94,iv:4gQ10k3Z3oZ2/pKrP530yWpaha16KznbwXJilLmJGO0=,tag:VvwHjDwFNRkPeBGauLtajw==,type:str]
+    instance_url: https://estuary2-dev-ed.develop.my.salesforce.com
 is_sandbox: false
 start_date: "2024-07-26T00:00:00Z"
 advanced:
@@ -16,8 +17,8 @@ sops:
     azure_kv: []
     hc_vault: []
     age: []
-    lastmodified: "2025-03-12T02:46:56Z"
-    mac: ENC[AES256_GCM,data:0NacPbDKrJcFItn8Pa0sh3pFUEcjubauI3dsLofT4csBqO5+VHg23mCMDcYB5oYGb3SK58RMn1r8Cx1WunChczQdoeWiE41ga2p92u8Vag5enZQomk7cqV/d6/eLb38KbV7YR1MkXg5XW3OorTkcVnqXfDUrIWOTMddM9WDQh8I=,iv:F7lYY8p5HoWUv6qSeKA14kwfA4LVLEksB0RqniMmwEw=,tag:n6/2qysw4FEQHhz60059uw==,type:str]
+    lastmodified: "2025-03-14T18:11:44Z"
+    mac: ENC[AES256_GCM,data:2q6JA+SFc4w+MsfeuapyfuVZq9n/ATOdIJxEurebzR2TDToOsfqhSR+c2rEM9aIErD9UDL2xc6hYtfu8hCg5YCdsyqAQXYxFAfASa5xWiHodAZ5H4k2jaY9Yf8Yi0UM1IVW2tt04o8qc2yrLut7mnVoQtjPfzT/782LjgfZZt6s=,iv:qoMbdN3i4rO1S/EoqEZ4gTggf89FPKkOh3qNZgjmbfs=,tag:HUKjKKjBojrETyyk5oNiLg==,type:str]
     pgp: []
     encrypted_suffix: _sops
     version: 3.7.3

source-salesforce-native/source_salesforce_native/__init__.py

@@ -19,36 +19,36 @@
     ConnectorState,
     EndpointConfig,
     OAUTH2_SPEC,
-    ResourceConfig,
+    SalesforceResourceConfigWithSchedule,
 )
 
 
 class Connector(
-    BaseCaptureConnector[EndpointConfig, ResourceConfig, ConnectorState],
+    BaseCaptureConnector[EndpointConfig, SalesforceResourceConfigWithSchedule, ConnectorState],
     HTTPMixin,
 ):
     def request_class(self):
-        return Request[EndpointConfig, ResourceConfig, ConnectorState]
+        return Request[EndpointConfig, SalesforceResourceConfigWithSchedule, ConnectorState]
 
     async def spec(self, _: request.Spec, logger: Logger) -> ConnectorSpec:
         return ConnectorSpec(
             configSchema=EndpointConfig.model_json_schema(),
             oauth2=OAUTH2_SPEC,
             documentationUrl="https://go.estuary.dev/source-salesforce-native",
-            resourceConfigSchema=ResourceConfig.model_json_schema(),
-            resourcePathPointers=ResourceConfig.PATH_POINTERS,
+            resourceConfigSchema=SalesforceResourceConfigWithSchedule.model_json_schema(),
+            resourcePathPointers=SalesforceResourceConfigWithSchedule.PATH_POINTERS,
         )
 
     async def discover(
         self, log: Logger, discover: request.Discover[EndpointConfig]
-    ) -> response.Discovered[ResourceConfig]:
+    ) -> response.Discovered[SalesforceResourceConfigWithSchedule]:
         resources = await all_resources(log, self, discover.config)
         return common.discovered(resources)
 
     async def validate(
         self,
         log: Logger,
-        validate: request.Validate[EndpointConfig, ResourceConfig],
+        validate: request.Validate[EndpointConfig, SalesforceResourceConfigWithSchedule],
     ) -> response.Validated:
         resources = await enabled_resources(log, self, validate.config, validate.bindings)
         resolved = common.resolve_bindings(validate.bindings, resources)
@@ -57,7 +57,7 @@ async def validate(
     async def open(
         self,
         log: Logger,
-        open: request.Open[EndpointConfig, ResourceConfig, ConnectorState],
+        open: request.Open[EndpointConfig, SalesforceResourceConfigWithSchedule, ConnectorState],
     ) -> tuple[response.Opened, Callable[[Task], Awaitable[None]]]:
         resources = await enabled_resources(log, self, open.capture.config, open.capture.bindings)
         resolved = common.resolve_bindings(open.capture.bindings, resources)

source-salesforce-native/source_salesforce_native/api.py

@@ -7,8 +7,9 @@
 
 from .bulk_job_manager import BulkJobError, BulkJobManager, NOT_SUPPORTED_BY_BULK_API, CANNOT_FETCH_COMPOUND_DATA
 from .rest_query_manager import RestQueryManager
-from .shared import dt_to_str, str_to_dt
+from .shared import dt_to_str, str_to_dt, now
 from .models import (
+    FieldDetails,
     FieldDetailsDict,
     FullRefreshResource,
     SalesforceResource,
@@ -34,6 +35,21 @@ def _determine_cursor_field(
         raise RuntimeError("Attempted to find cursor field but no valid cursor field exists.")
 
 
+def _filter_to_only_formula_fields(all_fields: FieldDetailsDict, cursor_field: str) -> tuple[bool, FieldDetailsDict]:
+    mandatory_fields: list[str] = ['Id', cursor_field]
+
+    mandatory_and_formula_fields: dict[str, FieldDetails] = {}
+    has_formula_fields = False
+    for field, details in all_fields.items():
+        if field in mandatory_fields or details.calculated:
+            mandatory_and_formula_fields[field] = details
+        
+        if details.calculated:
+            has_formula_fields = True
+
+    return (has_formula_fields, FieldDetailsDict.model_validate(mandatory_and_formula_fields))
+
+
 async def snapshot_resources(
     http: HTTPSession,
     bulk_job_manager: BulkJobManager,
@@ -93,6 +109,7 @@ async def backfill_incremental_resources(
     log: Logger,
     page: PageCursor | None,
     cutoff: LogCursor,
+    is_connector_initiated: bool,
 ) -> AsyncGenerator[SalesforceResource | PageCursor, None]:
     assert isinstance(page, str)
     assert isinstance(cutoff, datetime)
@@ -106,6 +123,16 @@ async def backfill_incremental_resources(
 
     cursor_field = _determine_cursor_field(fields)
 
+    # On connector-initiated backfills, only fetch formula fields and rely on the top level
+    # merge reduction strategy to merge in partial documents containing updated formula fields.
+    if is_connector_initiated:
+        log.info('ALEX - saw connector_initiated argument! only fetching formula fields')
+        has_formula_fields, fields = _filter_to_only_formula_fields(fields, cursor_field)
+        # If there are no formula fields in this object, return early.
+        if not has_formula_fields:
+            log.info("HEYA - there aren't any formula fields, finishing early")
+            return
+
     async def _execute(
         manager: BulkJobManager | RestQueryManager, 
         checkpoint_interval: int
@@ -153,7 +180,7 @@ async def fetch_incremental_resources(
 ) -> AsyncGenerator[SalesforceResource | LogCursor, None]:
     assert isinstance(log_cursor, datetime)
 
-    end = min(datetime.now(tz=UTC), log_cursor + timedelta(days=window_size))
+    end = min(now(), log_cursor + timedelta(days=window_size))
 
     cursor_field = _determine_cursor_field(fields)
 

source-salesforce-native/source_salesforce_native/models.py

@@ -9,7 +9,8 @@
     BaseDocument,
     BaseOAuth2Credentials,
     OAuth2Spec,
-    ResourceConfig,
+    ResourceConfigWithSchedule,
+    CRON_REGEX,
     ResourceState,
 )
 from estuary_cdk.capture.common import (
@@ -98,6 +99,15 @@ def _you_must_build_oauth2_credentials_for_a_provider(self): ...
     OAuth2Credentials = SalesforceOAuth2Credentials.for_provider(OAUTH2_SPEC.provider)
 
 
+class SalesforceResourceConfigWithSchedule(ResourceConfigWithSchedule):
+    schedule: str = Field(
+        default="",
+        title="Formula Field Refresh Schedule",
+        description="Schedule to automatically refresh formula fields. Accepts a cron expression.",
+        pattern=CRON_REGEX
+    )
+
+
 class EndpointConfig(BaseModel):
     start_date: AwareDatetime = Field(
         description="UTC data and time in the format YYYY-MM-DDTHH:MM:SSZ. Any data generated before this date will not be replicated. If left blank, all data will be replicated.",

source-salesforce-native/source_salesforce_native/resources.py

@@ -6,17 +6,18 @@
 
 from estuary_cdk.flow import CaptureBinding
 from estuary_cdk.capture import common, Task
+from estuary_cdk.capture.common import ReductionStrategy
 from estuary_cdk.http import HTTPMixin
 
 from .supported_standard_objects import SUPPORTED_STANDARD_OBJECTS, COMMON_CUSTOM_OBJECT_DETAILS
 
 from .bulk_job_manager import BulkJobManager
 from .rest_query_manager import RestQueryManager
-from .shared import dt_to_str, VERSION
+from .shared import dt_to_str, now, VERSION
 
 from .models import (
     EndpointConfig,
-    ResourceConfig,
+    SalesforceResourceConfigWithSchedule,
     ResourceState,
     SalesforceTokenSource,
     GlobalDescribeObjectsResponse,
@@ -79,7 +80,7 @@ def full_refresh_resource(
 ) -> common.Resource:
 
     def open(
-        binding: CaptureBinding[ResourceConfig],
+        binding: CaptureBinding[SalesforceResourceConfigWithSchedule],
         binding_index: int,
         state: ResourceState,
         task: Task,
@@ -112,7 +113,7 @@ def open(
         model=FullRefreshResource,
         open=open,
         initial_state=ResourceState(),
-        initial_config=ResourceConfig(
+        initial_config=SalesforceResourceConfigWithSchedule(
             name=name, interval=timedelta(minutes=5)
         ),
         schema_inference=True,
@@ -136,7 +137,7 @@ def incremental_resource(
 ) -> common.Resource:
 
     def open(
-        binding: CaptureBinding[ResourceConfig],
+        binding: CaptureBinding[SalesforceResourceConfigWithSchedule],
         binding_index: int,
         state: ResourceState,
         task: Task,
@@ -176,7 +177,7 @@ def open(
             ),
         )
 
-    cutoff = datetime.now(tz=UTC).replace(microsecond=0)
+    cutoff = now()
 
     resource = common.Resource(
         name=name,
@@ -187,11 +188,15 @@ def open(
             inc=ResourceState.Incremental(cursor=cutoff),
             backfill=ResourceState.Backfill(next_page=dt_to_str(config.start_date), cutoff=cutoff)
         ),
-        initial_config=ResourceConfig(
-            name=name, interval=timedelta(minutes=5)
+        initial_config=SalesforceResourceConfigWithSchedule(
+            name=name,
+            interval=timedelta(minutes=5),
+            # Default to performing a formula field refresh at 23:55 UTC every day for every enabled binding.
+            schedule="55 23 * * *"
         ),
         schema_inference=True,
         disable=not enable,
+        reduction_strategy=ReductionStrategy.MERGE,
     )
 
     return resource

source-salesforce-native/source_salesforce_native/shared.py

@@ -14,6 +14,12 @@ def str_to_dt(string: str) -> datetime:
     return datetime.fromisoformat(string)
 
 
+# Salesforce's datetimes have millisecond precision. now helps ensure
+# we are always working with millisecond precision datetimes.
+def now() -> datetime:
+    return str_to_dt(dt_to_str(datetime.now(tz=UTC)))
+
+
 def build_query(
         object_name: str, 
         fields: list[str],