From 0a0f1c34977553ac6353554184b99cb3c8fd394f Mon Sep 17 00:00:00 2001 From: Richard McDaniel <1130888+rmcdaniel@users.noreply.github.com> Date: Wed, 11 Nov 2026 23:51:35 +0000 Subject: [PATCH] Accept sticky history suffixes only when they continue the cache (#300) Workers with a sticky cache advertise sticky_history_suffix v1 (server#1017). A suffix answer is used only on a sticky hit whose cached history ends at exactly history_after_sequence and whose events run gap-free to history_last_sequence; anything else drops the entry and refetches full history from the start. The completion claim reports the cached last sequence when the Server supports suffixes. Co-Authored-By: Claude Opus 5.5 --- CHANGELOG.md | 4 + src/durable_workflow/worker.py | 74 ++++++++- .../sticky-history-suffix-gap-refetches.json | 48 ++++++ tests/test_replay_regression_corpus.py | 46 ++++++ tests/test_sticky_history_suffix.py | 145 ++++++++++++++++++ tests/test_worker.py | 42 +++++ 6 files changed, 357 insertions(+), 2 deletions(-) create mode 100644 tests/fixtures/replay_regressions/sticky-history-suffix-gap-refetches.json create mode 100644 tests/test_sticky_history_suffix.py diff --git a/CHANGELOG.md b/CHANGELOG.md index b6682c0..71b320b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -6,6 +6,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- Workers with a sticky cache (`sticky_cache_capacity > 0`) advertise `sticky_history_suffix` (#300, server#1017). On Server 3.5.0 or later, a sticky hit receives only the history after the sequence the worker cached, instead of the full history on every workflow task. A suffix is used only when it continues the cached history exactly; otherwise the worker drops the entry and refetches full history. + ## [2.15.2] - 2026-11-11 ### Fixed diff --git a/src/durable_workflow/worker.py b/src/durable_workflow/worker.py index 1ad5fb5..797a441 100644 --- a/src/durable_workflow/worker.py +++ b/src/durable_workflow/worker.py @@ -430,6 +430,18 @@ def _signal_arguments_envelope_from_export( return envelope +def _suffix_continues(events: list[dict[str, Any]], after: int, last: Any) -> bool: + """Whether ``events`` run gap-free from ``after + 1`` and end at ``last``.""" + if type(last) is not int or last < after: + return False + if any(not isinstance(event, dict) for event in events): + return False + sequences = [event.get("sequence") for event in events] + if any(type(sequence) is not int for sequence in sequences): + return False + return sequences == list(range(after + 1, last + 1)) + + def _history_events_from_query_export(history: Any, history_export: Any) -> Any: if isinstance(history, list) and history: return history @@ -1215,6 +1227,7 @@ def __init__( self._poller_tasks: set[asyncio.Task[Any]] = set() self._registered = False self._deregistration_fenced = False + self._sticky_history_suffix_supported = False self._registration_token: str | None = None self._registration_worker_id: str | None = None self._registration_namespace: object = None @@ -1383,6 +1396,14 @@ async def _register(self) -> None: not isinstance(sticky_support, Mapping) or sticky_support.get("supported") is not True ): raise RuntimeError("sticky execution requires Server's advertised sticky_execution support") + # Server #1017: a sticky worker may take only the history after its cached sequence. + suffix_support = ( + server_capabilities.get("sticky_history_suffix") if isinstance(server_capabilities, Mapping) else None + ) + self._sticky_history_suffix_supported = ( + self._sticky_cache.enabled and isinstance(suffix_support, Mapping) + and suffix_support.get("supported") is True and suffix_support.get("version") == 1 + ) self._cooperative_cancellation_supported = ( "cooperative_cancellation" in self.capabilities and isinstance(protocol, Mapping) @@ -1485,7 +1506,9 @@ async def _register(self) -> None: capabilities.append("prepared_local_activity_cancellation_policies") if PORTABLE_WORKER_AFFINITY_CAPABILITY_MANIFEST["worker_sessions"]["supported"]: capabilities.append("worker_sessions") - affinity_manifest = {name: dict(entry) for name, entry in PORTABLE_WORKER_AFFINITY_CAPABILITY_MANIFEST.items()} + affinity_manifest: dict[str, dict[str, Any]] = { + name: dict(entry) for name, entry in PORTABLE_WORKER_AFFINITY_CAPABILITY_MANIFEST.items() + } if self._sticky_cache.enabled: if "sticky_execution" not in capabilities: capabilities.append("sticky_execution") @@ -1494,6 +1517,11 @@ async def _register(self) -> None: "minimum_protocol_version": "1.18", "implementation": "bounded_durable_history_cache", } + if "sticky_history_suffix" not in capabilities: + capabilities.append("sticky_history_suffix") + affinity_manifest["sticky_history_suffix"] = { + "supported": True, "minimum_protocol_version": "1.18", "version": 1, + } ack = await self.client.register_worker( worker_id=self.worker_id, @@ -1683,6 +1711,9 @@ async def _load_workflow_claim_history( ) -> list[dict[str, Any]]: if first_page_token is not None: return await self._fetch_workflow_claim_history_pages(task, [], first_page_token) + after = task.get("history_after_sequence") + if type(after) is int and after > 0: + return await self._load_suffix_history(task, after) inline = list(task.get("history_events", [])) key = self._sticky_cache_key(task) mode = task.get("sticky_replay_mode") @@ -1735,6 +1766,41 @@ async def _load_workflow_claim_history( return history return await self._fetch_workflow_claim_history_pages(task, inline, task.get("next_history_page_token")) + async def _load_suffix_history(self, task: dict[str, Any], after: int) -> list[dict[str, Any]]: + """Replay a sticky_history_suffix v1 answer (server#1017) or refetch full history. + + A suffix is used only on a sticky hit whose cached history ends at exactly + ``after`` and whose events run gap-free from ``after + 1`` to + ``history_last_sequence``. It is never treated as a complete history. + """ + key = self._sticky_cache_key(task) + suffix = list(task.get("history_events", [])) + token = task.get("next_history_page_token") + if token is not None: + suffix = await self._fetch_workflow_claim_history_pages(task, suffix, token) + last = task.get("history_last_sequence") + cached = ( + self._sticky_cache.lookup(key) + if key is not None and self._sticky_history_suffix_supported + and task.get("sticky_replay_mode") == "sticky_hit_expected" else None + ) + if cached is not None: + prefix = cached[0] + if prefix and prefix[-1].get("sequence") == after and _suffix_continues(suffix, after, last): + history = prefix + suffix + if complete_history(history): + task["_sticky_resume_token"] = None + task["_sticky_resume_offset"] = 0 + self._sticky_cache.record_replay(hit=True) + return history + if key is not None: + self._sticky_cache.discard(key) + self._sticky_cache.record_replay(hit=False, forced=True) + history = await self._fetch_workflow_claim_history_pages(task, [], "MA==") + if not complete_history(history): + raise LocalActivityExecutionAborted("cold replay lacks a complete canonical history prefix") + return history + async def _fetch_workflow_claim_history_pages( self, task: dict[str, Any], history: list[dict[str, Any]], token: Any, ) -> list[dict[str, Any]]: @@ -1782,12 +1848,16 @@ def _sticky_cache_claim( resume_offset=task.get("_sticky_resume_offset", 0), ): return None - return { + claim: dict[str, Any] = { "worker_id": self.worker_id, "workflow_id": key[0], "run_id": key[1], "build_id": key[2], "ttl_seconds": self._sticky_cache.ttl_seconds, "metrics": {name: self._sticky_cache.metrics()[name] for name in ("hit", "miss", "eviction", "forced_cold_replay")}, } + if self._sticky_history_suffix_supported: + # sticky_history_suffix v1: the Server sends the next suffix after exactly this sequence. + claim["history_last_sequence"] = history[-1]["sequence"] + return claim def sticky_cache_metrics(self) -> dict[str, int]: """Return replay counters, retained entries and encoded history bytes.""" diff --git a/tests/fixtures/replay_regressions/sticky-history-suffix-gap-refetches.json b/tests/fixtures/replay_regressions/sticky-history-suffix-gap-refetches.json new file mode 100644 index 0000000..01e30b0 --- /dev/null +++ b/tests/fixtures/replay_regressions/sticky-history-suffix-gap-refetches.json @@ -0,0 +1,48 @@ +{ + "$schema": "https://raw.githubusercontent.com/durable-workflow/.github/main/regression-corpus/evidence-schema.json", + "fixture_schema": "durable-workflow.replay-regression/v1", + "id": "python-sticky-history-suffix-gap-refetches", + "protocol_version": "1.20", + "bindings": ["python"], + "workflow": { + "type": "sdk.sticky-suffix", + "input": [ + {"cached_through": 4, "after": 4, "last": 6, "suffix": [5, 6]}, + {"cached_through": 4, "after": 4, "last": 7, "suffix": [6, 7]} + ] + }, + "command_sequence": [ + { + "type": "sticky_suffix_history", + "history_sequences": [1, 2, 3, 4, 5, 6], + "authoritative_fetch_tokens": [], + "hit": 1, + "forced_cold_replay": 0 + }, + { + "type": "sticky_suffix_history", + "history_sequences": [1, 2, 3, 4, 5, 6, 7], + "authoritative_fetch_tokens": ["MA=="], + "hit": 0, + "forced_cold_replay": 1 + } + ], + "expected": { + "command_sequence": [ + { + "type": "sticky_suffix_history", + "history_sequences": [1, 2, 3, 4, 5, 6], + "authoritative_fetch_tokens": [], + "hit": 1, + "forced_cold_replay": 0 + }, + { + "type": "sticky_suffix_history", + "history_sequences": [1, 2, 3, 4, 5, 6, 7], + "authoritative_fetch_tokens": ["MA=="], + "hit": 0, + "forced_cold_replay": 1 + } + ] + } +} diff --git a/tests/test_replay_regression_corpus.py b/tests/test_replay_regression_corpus.py index 078d1b4..e16c59d 100644 --- a/tests/test_replay_regression_corpus.py +++ b/tests/test_replay_regression_corpus.py @@ -543,6 +543,46 @@ async def record_sleep(seconds: float) -> None: return asyncio.run(run()) +def _execute_sticky_suffix(inputs: Any) -> list[dict[str, Any]]: + """Feed sticky_history_suffix v1 answers (server#1017) to the real worker over a warm + cache and report what it replays and which authoritative pages it fetched.""" + assert isinstance(inputs, list) and inputs and all(isinstance(item, dict) for item in inputs) + + def events(sequences: list[int]) -> list[dict[str, Any]]: + return [{"id": f"event-{sequence}", "sequence": sequence, + "event_type": "WorkflowStarted" if sequence == 1 else "SideEffectRecorded", + "payload": {"value": {"number": sequence}}} for sequence in sequences] + + async def run(scenario: dict[str, Any]) -> dict[str, Any]: + last = scenario["last"] + client = AsyncMock(spec=Client) + client.workflow_task_history.return_value = { + "history_events": events(list(range(1, last + 1))), "next_history_page_token": None, + } + worker = Worker(client, task_queue="queue", worker_id="holder", build_id="build", sticky_cache_capacity=1) + worker._sticky_history_suffix_supported = True + assert worker._sticky_cache.remember( + ("workflow", "run", "build"), events(list(range(1, scenario["cached_through"] + 1))), + ) + history = await worker._load_workflow_claim_history({ + "task_id": "task", "workflow_id": "workflow", "run_id": "run", "workflow_task_attempt": 1, + "sticky_replay_mode": "sticky_hit_expected", "history_events": events(scenario["suffix"]), + "history_after_sequence": scenario["after"], "history_last_sequence": last, + "next_history_page_token": None, + }) + metrics = worker.sticky_cache_metrics() + return { + "type": "sticky_suffix_history", + "history_sequences": [event["sequence"] for event in history], + "authoritative_fetch_tokens": [ + call.kwargs["next_history_page_token"] for call in client.workflow_task_history.await_args_list + ], + "hit": metrics["hit"], + "forced_cold_replay": metrics["forced_cold_replay"], + } + + return [asyncio.run(run(scenario)) for scenario in inputs] + def _execute_fixture(fixture: dict[str, Any]) -> list[dict[str, Any]]: assert fixture.get("fixture_schema") == FIXTURE_SCHEMA assert "python" in fixture.get("bindings", []) @@ -553,6 +593,11 @@ def _execute_fixture(fixture: dict[str, Any]) -> list[dict[str, Any]]: assert isinstance(workflow_type, str) and workflow_type if workflow_type == "sdk.observed-worker-replay": return _execute_observed_worker_replay(fixture) + if workflow_type == "sdk.sticky-suffix": + commands = _execute_sticky_suffix(workflow.get("input")) + _assert_matches(fixture["command_sequence"], commands, f"{fixture['id']}.command_sequence") + _assert_matches(fixture["expected"], {"command_sequence": commands}, f"{fixture['id']}.expected") + return commands if workflow_type == "sdk.local-activity-backoff": commands = [_execute_local_activity_backoff(workflow.get("input"))] _assert_matches(fixture["command_sequence"], commands, f"{fixture['id']}.command_sequence") @@ -782,6 +827,7 @@ def test_checked_in_replay_regression_corpus_uses_official_replayer( if workflow.get("type") in { "sdk.replay-conformance-report", "sdk.query-task-dispatch", "sdk.generator-query-task-dispatch", "sdk.generator-side-effect-query-task-dispatch", "sdk.observed-worker-replay", "sdk.local-activity-backoff", + "sdk.sticky-suffix", }: _execute_fixture(fixture) return diff --git a/tests/test_sticky_history_suffix.py b/tests/test_sticky_history_suffix.py new file mode 100644 index 0000000..087e3b5 --- /dev/null +++ b/tests/test_sticky_history_suffix.py @@ -0,0 +1,145 @@ +"""sticky_history_suffix v1 (server#1017): accept a suffix only when it continues the cache.""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import AsyncMock + +import pytest + +from durable_workflow import Client, Worker + +KEY = ("workflow", "run", "build") + + +def history(count: int) -> list[dict[str, Any]]: + return [{"id": f"event-{sequence}", "sequence": sequence, + "event_type": "WorkflowStarted" if sequence == 1 else "SideEffectRecorded", + "payload": {"value": {"number": sequence}}} for sequence in range(1, count + 1)] + + +def suffix_task(events: list[dict[str, Any]], *, after: int, last: int, + mode: str = "sticky_hit_expected", token: str | None = None) -> dict[str, Any]: + return {"task_id": "current-task", "workflow_id": "workflow", "run_id": "run", + "workflow_task_attempt": 2, "history_events": events, "sticky_replay_mode": mode, + "history_after_sequence": after, "history_last_sequence": last, + "last_history_sequence": last, "next_history_page_token": token} + + +def warm_worker(client: Any, cached: int = 4) -> Worker: + worker = Worker(client, task_queue="q", worker_id="holder", build_id="build", sticky_cache_capacity=2) + worker._sticky_history_suffix_supported = True + assert worker._sticky_cache.remember(KEY, history(cached)) + return worker + + +def cold_reply(count: int) -> dict[str, Any]: + return {"history_events": history(count), "next_history_page_token": None} + + +@pytest.mark.asyncio +async def test_hit_appends_the_suffix_to_the_cached_prefix() -> None: + client = AsyncMock(spec=Client) + worker = warm_worker(client) + loaded = await worker._load_workflow_claim_history(suffix_task(history(6)[4:], after=4, last=6)) + assert loaded == history(6) + client.workflow_task_history.assert_not_awaited() + assert worker.sticky_cache_metrics()["hit"] == 1 + assert worker.sticky_cache_metrics()["forced_cold_replay"] == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize(("events", "after", "last"), [ + pytest.param(history(7)[5:], 4, 7, id="gap"), + pytest.param(history(6)[3:], 4, 6, id="overlap"), + pytest.param([history(7)[4], history(7)[6]], 4, 7, id="non-contiguous"), + pytest.param(history(6)[4:5], 4, 6, id="ends-short-of-last"), + pytest.param(history(6)[4:], 3, 6, id="boundary-differs-from-cache"), +]) +async def test_suffix_that_does_not_continue_the_cache_refetches_full_history( + events: list[dict[str, Any]], after: int, last: int, +) -> None: + client = AsyncMock(spec=Client) + client.workflow_task_history.return_value = cold_reply(last) + worker = warm_worker(client) + loaded = await worker._load_workflow_claim_history(suffix_task(events, after=after, last=last)) + assert loaded == history(last) + client.workflow_task_history.assert_awaited_once_with( + task_id="current-task", next_history_page_token="MA==", lease_owner="holder", workflow_task_attempt=2, + ) + assert worker.sticky_cache_metrics()["hit"] == 0 + assert worker.sticky_cache_metrics()["forced_cold_replay"] == 1 + assert worker._sticky_cache.lookup(KEY) is None + + +@pytest.mark.asyncio +async def test_cold_suffix_without_a_cached_prefix_is_never_replayed_as_full_history() -> None: + client = AsyncMock(spec=Client) + client.workflow_task_history.return_value = cold_reply(6) + worker = Worker(client, task_queue="q", worker_id="replacement", build_id="build", sticky_cache_capacity=1) + worker._sticky_history_suffix_supported = True + assert await worker._load_workflow_claim_history(suffix_task(history(6)[4:], after=4, last=6)) == history(6) + assert client.workflow_task_history.await_args.kwargs["next_history_page_token"] == "MA==" + + +@pytest.mark.asyncio +@pytest.mark.parametrize("mode", ["forced_cold_replay", "cold_replay", None]) +async def test_suffix_outside_a_sticky_hit_refetches(mode: str | None) -> None: + client = AsyncMock(spec=Client) + client.workflow_task_history.return_value = cold_reply(6) + worker = warm_worker(client) + task = suffix_task(history(6)[4:], after=4, last=6) + task["sticky_replay_mode"] = mode + assert await worker._load_workflow_claim_history(task) == history(6) + assert client.workflow_task_history.await_args.kwargs["next_history_page_token"] == "MA==" + + +@pytest.mark.asyncio +async def test_empty_suffix_is_a_hit_only_when_nothing_was_appended() -> None: + client = AsyncMock(spec=Client) + worker = warm_worker(client) + assert await worker._load_workflow_claim_history(suffix_task([], after=4, last=4)) == history(4) + client.workflow_task_history.return_value = cold_reply(5) + assert await worker._load_workflow_claim_history(suffix_task([], after=4, last=5)) == history(5) + assert worker.sticky_cache_metrics()["hit"] == 1 + assert worker.sticky_cache_metrics()["forced_cold_replay"] == 1 + + +@pytest.mark.asyncio +async def test_paged_suffix_is_assembled_before_it_is_checked() -> None: + client = AsyncMock(spec=Client) + client.workflow_task_history.return_value = {"history_events": history(8)[6:], "next_history_page_token": None} + worker = warm_worker(client) + task = suffix_task(history(8)[4:6], after=4, last=8, token="suffix-page-two") + assert await worker._load_workflow_claim_history(task) == history(8) + assert client.workflow_task_history.await_args.kwargs["next_history_page_token"] == "suffix-page-two" + assert worker.sticky_cache_metrics()["hit"] == 1 + + +@pytest.mark.asyncio +async def test_server_without_suffix_support_never_uses_a_suffix() -> None: + client = AsyncMock(spec=Client) + client.workflow_task_history.return_value = cold_reply(6) + worker = warm_worker(client) + worker._sticky_history_suffix_supported = False + assert await worker._load_workflow_claim_history(suffix_task(history(6)[4:], after=4, last=6)) == history(6) + assert client.workflow_task_history.await_args.kwargs["next_history_page_token"] == "MA==" + + +@pytest.mark.asyncio +async def test_boundary_zero_keeps_the_full_history_path() -> None: + client = AsyncMock(spec=Client) + worker = warm_worker(client) + task = suffix_task(history(6), after=0, last=6, mode="cold_replay") + assert await worker._load_workflow_claim_history(task) == history(6) + client.workflow_task_history.assert_not_awaited() + + +def test_claim_reports_the_cached_last_sequence_only_with_server_support() -> None: + worker = Worker(AsyncMock(spec=Client), task_queue="q", build_id="build", sticky_cache_capacity=2) + task = suffix_task(history(6), after=0, last=6) + claim = worker._sticky_cache_claim(task, history(6), [{"type": "start_timer"}]) + assert claim is not None and "history_last_sequence" not in claim + worker._sticky_history_suffix_supported = True + claim = worker._sticky_cache_claim(task, history(6), [{"type": "start_timer"}]) + assert claim is not None and claim["history_last_sequence"] == 6 diff --git a/tests/test_worker.py b/tests/test_worker.py index 281af79..d8d575f 100644 --- a/tests/test_worker.py +++ b/tests/test_worker.py @@ -7,6 +7,7 @@ import sys import threading from pathlib import Path +from typing import Any from unittest.mock import AsyncMock, Mock import httpx @@ -1253,6 +1254,47 @@ async def test_only_enabled_worker_advertises_sticky_support(self, mock_client: } await worker.stop() + @pytest.mark.asyncio + @pytest.mark.parametrize(("server_entry", "expected"), [ + ({"supported": True, "version": 1}, True), + ({"supported": True, "version": 2}, False), + ({"supported": False, "version": 1}, False), + (None, False), + ]) + async def test_enabled_worker_advertises_history_suffix_and_reads_server_support( + self, mock_client: AsyncMock, server_entry: dict[str, Any] | None, expected: bool, + ) -> None: + info = compatible_cluster_info() + capabilities = info["worker_protocol"]["server_capabilities"] # type: ignore[index] + capabilities["sticky_execution"] = {"supported": True} + if server_entry is not None: + capabilities["sticky_history_suffix"] = server_entry + mock_client.get_cluster_info.return_value = info + worker = Worker(mock_client, task_queue="q", sticky_cache_capacity=2) + await worker._register() + registration = mock_client.register_worker.await_args.kwargs + assert registration["capabilities"].count("sticky_history_suffix") == 1 + assert registration["capability_manifest"]["sticky_history_suffix"] == { + "supported": True, "minimum_protocol_version": "1.18", "version": 1, + } + assert worker._sticky_history_suffix_supported is expected + await worker.stop() + + @pytest.mark.asyncio + async def test_disabled_cache_does_not_advertise_history_suffix(self, mock_client: AsyncMock) -> None: + info = compatible_cluster_info() + info["worker_protocol"]["server_capabilities"]["sticky_history_suffix"] = { # type: ignore[index] + "supported": True, "version": 1, + } + mock_client.get_cluster_info.return_value = info + worker = Worker(mock_client, task_queue="q") + await worker._register() + registration = mock_client.register_worker.await_args.kwargs + assert "sticky_history_suffix" not in registration["capabilities"] + assert "sticky_history_suffix" not in registration["capability_manifest"] + assert worker._sticky_history_suffix_supported is False + await worker.stop() + @pytest.mark.asyncio async def test_enabled_cache_requires_server_support(self, mock_client: AsyncMock) -> None: worker = Worker(mock_client, task_queue="q", sticky_cache_capacity=2)