Skip unused scheduled activity inputs during Functions replay

Author: andystaples

azure-functions-durable/CHANGELOG.md

@@ -16,7 +16,8 @@ inputs and outputs, custom status, external events, entity state and results,
 and registered blueprints. History APIs also hydrate entity operation inputs
 and results. Registered orchestration and entity handlers await asynchronous
 payload storage without occupying synchronous execution threads. Replay does
-not download unused nested inputs from historical entity requests.
+not download unused inputs from historical scheduled activities or nested inputs
+from historical entity requests. Explicit history retrieval still hydrates them.
 User orchestrators remain generators, and activities retain their synchronous
 or asynchronous calling convention. Synchronous functions retain synchronous
 client bindings, and invocation logging context is preserved.

azure-functions-durable/azure/durable_functions/internal/payloads.py

@@ -224,6 +224,13 @@ def _entity_request_events(request: OrchestratorRequest) -> list[tuple[HistoryEv
     ]
 
 
+def discard_scheduled_activity_inputs(request: OrchestratorRequest) -> None:
+    """Omit activity inputs that replay never consumes before downloading references."""
+    for event in chain(request.pastEvents, request.newEvents):
+        if event.HasField("taskScheduled"):
+            event.taskScheduled.ClearField("input")
+
+
 def _update_entity_request_events(events: list[tuple[HistoryEvent, history.HistoryEvent]]) -> None:
     for source, event in events:
         if isinstance(event, history.EventRaisedEvent) and event.input is not None:

azure-functions-durable/azure/durable_functions/worker.py

@@ -23,7 +23,10 @@
 from .internal.invocation import run_sync
 from .internal.compat.entity_context import wrap_entity
 from .internal.compat.orchestration_context import wrap_orchestrator
-from .internal.payloads import get_transport_payload_store, hydrate_entity_request, hydrate_entity_request_async
+from .internal.payloads import (
+    discard_scheduled_activity_inputs, get_transport_payload_store,
+    hydrate_entity_request, hydrate_entity_request_async,
+)
 from .internal.serialization import DEFAULT_FUNCTIONS_DATA_CONVERTER
 
 _LOGGER = logging.getLogger(__name__)
@@ -98,6 +101,7 @@ def execute_orchestration_request(self, func: task.Orchestrator[Any, Any], conte
         request = OrchestratorRequest()
         request.ParseFromString(base64.b64decode(orchestration_context))
         if payload_store is not None:
+            discard_scheduled_activity_inputs(request)
             deexternalize_payloads(request, payload_store)
             hydrate_entity_request(request, payload_store)
         response = self._run_orchestration(func, request)
@@ -110,6 +114,7 @@ async def execute_orchestration_request_async(self, func: task.Orchestrator[Any,
         request = OrchestratorRequest()
         request.ParseFromString(base64.b64decode(getattr(context, "body", None) or context))
         if payload_store is not None:
+            discard_scheduled_activity_inputs(request)
             await deexternalize_payloads_async(request, payload_store)
             await hydrate_entity_request_async(request, payload_store)
         response = await run_sync(self._run_orchestration, func, request)

tests/azure-functions-durable/test_client_compat.py

@@ -161,7 +161,10 @@ async def test_history_hydrates_only_correlated_entity_envelopes(
     ordinary = pb.HistoryEvent(eventId=3)
     ordinary.eventRaised.name = "application-event"
     ordinary.eventRaised.input.value = reply.eventRaised.input.value
-    chunks = [pb.HistoryChunk(events=[request]), pb.HistoryChunk(events=[reply, ordinary])]
+    scheduled = pb.HistoryEvent(eventId=4)
+    scheduled.taskScheduled.name = "echo"
+    scheduled.taskScheduled.input.value = token
+    chunks = [pb.HistoryChunk(events=[request]), pb.HistoryChunk(events=[reply, ordinary, scheduled])]
 
     async def stream():
         for chunk in chunks:
@@ -180,11 +183,12 @@ async def stream():
             events = client.get_orchestration_history("instance", execution_id="execution")
         assert json.loads(json.loads(events[1].input)["result"]) == {"data": "hydrated"}
         assert events[2].input == ordinary.eventRaised.input.value
+        assert json.loads(events[3].input) == {"data": "hydrated"}
         assert json.loads(reply.eventRaised.input.value)["result"] == token
         if not modern_request:
             assert json.loads(json.loads(events[0].input)["input"]) == {"data": "hydrated"}
         assert stub.StreamInstanceHistory.call_args.args[0].executionId.value == "execution"
-        expected_downloads = 1 if modern_request else 2
+        expected_downloads = 2 if modern_request else 3
         assert store.download.call_count == (0 if use_async else expected_downloads)
         assert store.download_async.await_count == (expected_downloads if use_async else 0)
     finally:

tests/azure-functions-durable/test_worker_compat.py

@@ -353,6 +353,48 @@ def orchestrator(context, value):
         store.download_async.assert_not_called()
 
 
+@pytest.mark.parametrize("use_async", [False, True])
+@pytest.mark.parametrize("missing_blobs", [False, True])
+async def test_replay_skips_scheduled_inputs(monkeypatch, payload_store_factory, use_async, missing_blobs):
+    store = payload_store_factory()
+    monkeypatch.setattr(payloads, "_payload_store", store)
+    request = pb.OrchestratorRequest(instanceId=TEST_INSTANCE_ID)
+    request.pastEvents.extend([
+        helpers.new_orchestrator_started_event(),
+        helpers.new_execution_started_event("activity-replay", TEST_INSTANCE_ID),
+    ])
+    for task_id in range(1, 21):
+        token = store.upload(json.dumps("x" * 300_000).encode())
+        events = request.pastEvents if task_id < 20 else request.newEvents
+        events.extend([
+            helpers.new_orchestrator_started_event(),
+            helpers.new_task_scheduled_event(task_id, "echo", encoded_input=json.dumps(token)),
+            helpers.new_task_completed_event(task_id, json.dumps("ok")),
+        ])
+    if missing_blobs:
+        store._blobs.clear()
+    download = store.download
+    monkeypatch.setattr(store, "download", Mock(side_effect=download))
+    monkeypatch.setattr(store, "download_async", AsyncMock(side_effect=download))
+
+    def orchestrator(context, value):
+        for index in range(20):
+            result = yield context.call_activity("echo", input="x" * 300_000)
+            assert result == "ok"
+        return "complete"
+
+    worker = DurableFunctionsWorker()
+    encoded = base64.b64encode(request.SerializeToString()).decode()
+    result = (await worker.execute_orchestration_request_async(orchestrator, encoded) if use_async
+              else worker.execute_orchestration_request(orchestrator, encoded))
+    completion = _get_completion_action(_decode_orchestrator_response(result))
+    assert completion.orchestrationStatus == pb.ORCHESTRATION_STATUS_COMPLETED
+    assert json.loads(completion.result.value) == "complete"
+    store.download.assert_not_called()
+    store.download_async.assert_not_called()
+    assert request.pastEvents[3].taskScheduled.HasField("input")
+
+
 def _encode_orchestrator_request(name, encoded_input=None, instance_id=TEST_INSTANCE_ID):
     """Build a base64-encoded ``OrchestratorRequest`` for a single new dispatch."""
     request = pb.OrchestratorRequest(instanceId=instance_id)