estuary-cdk: incremental tasks are idle if their cursor is very recent

Some resources always have _some_ level of activity every time they're
checked, which would ordinarily prevent the resource from ever being
considered "caught up", preventing the connector from exiting.

If the LogCursor is a date-time, then additionally consider the task to
be idle if the LogCursor is less-than `interval` old (where interval is
configured on the ResourceConfig), and sleep until the cursor is
`interval` old before issuing a successive fetch.

estuary-cdk/estuary_cdk/capture/common.py

@@ -598,15 +598,32 @@ async def _binding_incremental_task(
                 "Implementation error: FetchChangesFn yielded a documents without a final LogCursor",
             )
 
-        if checkpoints:
-            continue  # Immediately fetch subsequent changes.
+        sleep_for : timedelta = binding.resourceConfig.interval
+
+        if not checkpoints:
+            # We're idle. Sleep for the full back-off interval.
+            sleep_for = binding.resourceConfig.interval 
+
+        elif isinstance(state.cursor, datetime):
+            lag = (datetime.now(tz=UTC) - state.cursor)
+
+            if lag > binding.resourceConfig.interval:
+                # We're not idle. Attempt to fetch the next changes.
+                continue
+            else:
+                # We're idle. Sleep until the cursor is `interval` old.
+                sleep_for = binding.resourceConfig.interval - lag
+        else:
+            # We're not idle. Attempt to fetch the next changes.
+            continue
+
+        task.log.debug("incremental task is idle", {"sleep_for": sleep_for, "cursor": state.cursor})
 
         # At this point we've fully caught up with the log and are idle.
         try:
             if not task.stopping.event.is_set():
                 await asyncio.wait_for(
-                    task.stopping.event.wait(),
-                    timeout=binding.resourceConfig.interval.total_seconds(),
+                    task.stopping.event.wait(), timeout=sleep_for.total_seconds()
                 )
 
             task.log.debug(f"incremental replication is idle and is yielding to stop")