Apply payload warnings to schedule inputs

durable-workflow-ops · durable-workflow-ops · commit cb19627ae7fb · 2026-04-21T17:03:14.000Z
Issue: zorporation/durable-workflow#444

Loop-ID: build-01
diff --git a/README.md b/README.md
@@ -156,7 +156,7 @@ restart the worker process with a new id before serving changed workflow code.
 - **Polyglot**: Works alongside PHP workers on the same task queue
 - **HTTP/JSON protocol**: No gRPC, no protobuf dependencies
 - **Codec envelopes**: Avro payloads by default, with JSON decode compatibility for existing history
-- **Payload-size warnings**: Structured warnings before oversized workflow, activity, signal, update, query, or search-attribute payloads reach the server
+- **Payload-size warnings**: Structured warnings before oversized workflow, activity, schedule, signal, update, query, or search-attribute payloads reach the server
 - **Workflow definition guard**: Worker registration refuses same-id hot reloads when a workflow class definition changed
 - **Worker interceptors**: Typed hooks around workflow tasks, activity calls, and query tasks for tracing, logging, and custom metrics
 - **Metrics hooks**: Pluggable counters and histograms, with an optional Prometheus adapter
@@ -165,9 +165,9 @@ restart the worker process with a new id before serving changed workflow code.
 
 The SDK logs a structured warning before an encoded payload reaches 80% of the
 default 2 MiB server payload limit. Warnings include context such as
-`workflow_id`, `activity_name`, `signal_name`, `update_name`, `query_name`,
-`payload_size`, `threshold_bytes`, and `limit_bytes` when those fields are
-known at the call site.
+`workflow_id`, `workflow_type`, `activity_name`, `schedule_id`, `signal_name`,
+`update_name`, `query_name`, `payload_size`, `threshold_bytes`, and
+`limit_bytes` when those fields are known at the call site.
 
 Tune or disable the warning threshold on the client:
 
diff --git a/src/durable_workflow/client.py b/src/durable_workflow/client.py
@@ -20,6 +20,7 @@
 
 import asyncio
 import time
+from collections.abc import Callable
 from dataclasses import dataclass
 from importlib.metadata import PackageNotFoundError
 from importlib.metadata import version as _pkg_version
@@ -130,12 +131,16 @@ class ScheduleAction:
     execution_timeout_seconds: int | None = None
     run_timeout_seconds: int | None = None
 
-    def to_dict(self) -> dict[str, Any]:
+    def to_dict(
+        self,
+        *,
+        input_encoder: Callable[[list[Any]], dict[str, str]] | None = None,
+    ) -> dict[str, Any]:
         d: dict[str, Any] = {"workflow_type": self.workflow_type}
         if self.task_queue is not None:
             d["task_queue"] = self.task_queue
         if self.input is not None:
-            d["input"] = serializer.envelope(self.input)
+            d["input"] = input_encoder(self.input) if input_encoder else serializer.envelope(self.input)
         if self.execution_timeout_seconds is not None:
             d["execution_timeout_seconds"] = self.execution_timeout_seconds
         if self.run_timeout_seconds is not None:
@@ -392,6 +397,7 @@ def _payload_context(
         *,
         kind: str,
         workflow_id: str | None = None,
+        workflow_type: str | None = None,
         run_id: str | None = None,
         activity_name: str | None = None,
         signal_name: str | None = None,
@@ -403,6 +409,7 @@ def _payload_context(
         return serializer.PayloadSizeWarningContext(
             kind=kind,
             workflow_id=workflow_id,
+            workflow_type=workflow_type,
             run_id=run_id,
             activity_name=activity_name,
             signal_name=signal_name,
@@ -420,6 +427,7 @@ def _payload_envelope(
         kind: str,
         codec: str = serializer.AVRO_CODEC,
         workflow_id: str | None = None,
+        workflow_type: str | None = None,
         run_id: str | None = None,
         activity_name: str | None = None,
         signal_name: str | None = None,
@@ -435,6 +443,7 @@ def _payload_envelope(
             warning_context=self._payload_context(
                 kind=kind,
                 workflow_id=workflow_id,
+                workflow_type=workflow_type,
                 run_id=run_id,
                 activity_name=activity_name,
                 signal_name=signal_name,
@@ -901,7 +910,15 @@ async def create_schedule(
         """
         body: dict[str, Any] = {
             "spec": spec.to_dict(),
-            "action": action.to_dict(),
+            "action": action.to_dict(
+                input_encoder=lambda value: self._payload_envelope(
+                    value,
+                    kind="schedule_input",
+                    workflow_type=action.workflow_type,
+                    schedule_id=schedule_id,
+                    task_queue=action.task_queue,
+                )
+            ),
         }
         if schedule_id is not None:
             body["schedule_id"] = schedule_id
@@ -999,7 +1016,15 @@ async def update_schedule(
         if spec is not None:
             body["spec"] = spec.to_dict()
         if action is not None:
-            body["action"] = action.to_dict()
+            body["action"] = action.to_dict(
+                input_encoder=lambda value: self._payload_envelope(
+                    value,
+                    kind="schedule_input",
+                    workflow_type=action.workflow_type,
+                    schedule_id=schedule_id,
+                    task_queue=action.task_queue,
+                )
+            )
         if overlap_policy is not None:
             body["overlap_policy"] = overlap_policy
         if jitter_seconds is not None:
diff --git a/src/durable_workflow/serializer.py b/src/durable_workflow/serializer.py
@@ -54,6 +54,7 @@ class PayloadSizeWarningContext:
 
     kind: str
     workflow_id: str | None = None
+    workflow_type: str | None = None
     run_id: str | None = None
     activity_name: str | None = None
     signal_name: str | None = None
@@ -67,6 +68,7 @@ def to_log_context(self) -> dict[str, str]:
         values = {
             "kind": self.kind,
             "workflow_id": self.workflow_id,
+            "workflow_type": self.workflow_type,
             "run_id": self.run_id,
             "activity_name": self.activity_name,
             "signal_name": self.signal_name,
diff --git a/tests/test_schedules.py b/tests/test_schedules.py
@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import json
+import logging
 from unittest.mock import AsyncMock, patch
 
 import httpx
@@ -112,6 +113,59 @@ async def test_action_input_uses_codec_envelope(self, client: Client) -> None:
             assert action_input["codec"] == "avro"
             assert serializer.decode(action_input["blob"], codec="avro") == ["Alice", 42]
 
+    @pytest.mark.asyncio
+    async def test_action_input_warning_uses_client_policy(
+        self, client: Client, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        client.payload_size_warning_config = serializer.PayloadSizeWarningConfig(
+            limit_bytes=10,
+            threshold_percent=50,
+        )
+        resp = _mock_response(201, {"schedule_id": "sched-large", "outcome": "created"})
+        with (
+            patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp),
+            caplog.at_level(logging.WARNING, logger="durable_workflow.serializer"),
+        ):
+            await client.create_schedule(
+                schedule_id="sched-large",
+                spec=ScheduleSpec(cron_expressions=["0 * * * *"]),
+                action=ScheduleAction(
+                    workflow_type="greeter",
+                    task_queue="q1",
+                    input=["this payload is intentionally large"],
+                ),
+            )
+
+        payload = caplog.records[0].durable_workflow_payload
+        assert payload["kind"] == "schedule_input"
+        assert payload["workflow_type"] == "greeter"
+        assert payload["schedule_id"] == "sched-large"
+        assert payload["task_queue"] == "q1"
+        assert payload["namespace"] == "ns1"
+        assert payload["threshold_bytes"] == 5
+
+    @pytest.mark.asyncio
+    async def test_action_input_warning_can_be_disabled(
+        self, client: Client, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        client.payload_size_warning_config = None
+        resp = _mock_response(201, {"schedule_id": "sched-quiet", "outcome": "created"})
+        with (
+            patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp),
+            caplog.at_level(logging.WARNING, logger="durable_workflow.serializer"),
+        ):
+            await client.create_schedule(
+                schedule_id="sched-quiet",
+                spec=ScheduleSpec(cron_expressions=["0 * * * *"]),
+                action=ScheduleAction(
+                    workflow_type="greeter",
+                    task_queue="q1",
+                    input=["this payload is intentionally large"],
+                ),
+            )
+
+        assert caplog.records == []
+
     @pytest.mark.asyncio
     async def test_minimal(self, client: Client) -> None:
         resp = _mock_response(201, {"schedule_id": "auto-id", "outcome": "created"})
@@ -255,6 +309,34 @@ async def test_update_note(self, client: Client) -> None:
             body = mock.call_args.kwargs.get("json") or mock.call_args[1].get("json")
             assert body["note"] == "Updated note"
 
+    @pytest.mark.asyncio
+    async def test_update_action_input_warning_uses_schedule_id(
+        self, client: Client, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        client.payload_size_warning_config = serializer.PayloadSizeWarningConfig(
+            limit_bytes=10,
+            threshold_percent=50,
+        )
+        resp = _mock_response(200, {"schedule_id": "sched-1", "outcome": "updated"})
+        with (
+            patch.object(client._http, "request", new_callable=AsyncMock, return_value=resp),
+            caplog.at_level(logging.WARNING, logger="durable_workflow.serializer"),
+        ):
+            await client.update_schedule(
+                "sched-1",
+                action=ScheduleAction(
+                    workflow_type="ticker",
+                    task_queue="q2",
+                    input=["this update payload is intentionally large"],
+                ),
+            )
+
+        payload = caplog.records[0].durable_workflow_payload
+        assert payload["kind"] == "schedule_input"
+        assert payload["workflow_type"] == "ticker"
+        assert payload["schedule_id"] == "sched-1"
+        assert payload["task_queue"] == "q2"
+
     @pytest.mark.asyncio
     async def test_not_found(self, client: Client) -> None:
         resp = _mock_response(404, {"reason": "schedule_not_found", "message": "not found"})
