Warn when Durable Functions dependencies allow major upgrades

azure-functions-durable/CHANGELOG.md

@@ -9,6 +9,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ADDED
 
+- A best-effort background check warns when a Function App's requirements allow
+a future major version, which may introduce replay-breaking changes.
+
 - Added opt-in `HttpRetryOptions` on `context.call_http()` for durable initial
 request retries with configurable timing and status selection. Empty or omitted
 status lists select all non-2xx responses; exhausted retries fail the call.

azure-functions-durable/README.md

@@ -17,9 +17,34 @@ built on top of the [`durabletask`](https://pypi.org/project/durabletask/) SDK.
 ## Installation
 
 ```bash
-pip install azure-functions-durable
+pip install "azure-functions-durable>=2.0.0b1,<3"
 ```
 
+For deployments, also constrain `azure-functions-durable` in the Function
+App's `requirements.txt`:
+
+```text
+azure-functions-durable>=2.0.0b1,<3
+```
+
+Future major versions may include broad, replay-breaking changes. An automatic
+upgrade during deployment can cause existing orchestrations to fail
+unexpectedly. Review the [migration guide](MIGRATION_GUIDE.md) and test replay
+before upgrading. A major-version constraint does not pin the `durabletask`
+dependency or guarantee replay compatibility; lock the complete dependency
+graph for reproducible deployments.
+
+After the first durable-client, orchestrator, or entity invocation,
+each Python worker checks accessible `requirements.txt` files in the
+background and writes a host worker log warning when the declared constraints
+permit a later major version. The check follows local `-r` and `-c` files and
+applicable environment markers. It is best-effort and may miss uncommon version
+ranges; missing or unsupported requirements are treated as unknown. A
+lockfile's presence does not indicate whether deployment uses it, so the warning
+describes the risk of installing directly from the requirements. The check does
+not read dependency files during app startup or wait for the check before
+returning from an invocation.
+
 ## Overview
 
 Author orchestrations, activities, and entities as Azure Functions and let the

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

@@ -0,0 +1,205 @@
+# Copyright (c) Microsoft Corporation.
+# Licensed under the MIT License.
+
+"""Best-effort check for Function App dependencies that allow a new major."""
+
+import os
+import re
+import shlex
+from importlib.metadata import PackageNotFoundError, version
+from logging import Logger
+from pathlib import Path
+
+from packaging.markers import UndefinedComparison, UndefinedEnvironmentName
+from packaging.requirements import InvalidRequirement, Requirement
+from packaging.specifiers import SpecifierSet
+from packaging.utils import canonicalize_name
+from packaging.version import InvalidVersion, Version
+
+_PACKAGE = "azure-functions-durable"
+_INDEX_OPTIONS_WITH_VALUE = {
+    "-i", "-f", "--index-url", "--extra-index-url", "--find-links",
+    "--trusted-host", "--only-binary", "--no-binary",
+}
+_INDEX_FLAGS = {"--no-index", "--pre", "--prefer-binary"}
+
+
+def _read_specifiers(root: Path, logger: Logger) -> tuple[SpecifierSet, bool] | None:
+    specifiers: list[str] = []
+    required = False
+    allow_prereleases = False
+    active: set[tuple[Path, bool]] = set()
+    visited: set[tuple[Path, bool]] = set()
+
+    def read_file(path: Path, constraint: bool) -> bool:
+        nonlocal allow_prereleases, required
+        try:
+            path = path.resolve()
+            key = (path, constraint)
+            if key in active:
+                logger.debug("Cannot check Durable Functions dependencies: cyclic requirements in %s.", path)
+                return False
+            if key in visited:
+                return True
+            content = path.read_text(encoding="utf-8-sig")
+        except (OSError, UnicodeError) as exc:
+            logger.debug("Cannot read Function App requirements from %s: %s", path, exc)
+            return False
+
+        active.add(key)
+        try:
+            for raw_line in content.splitlines():
+                line = re.sub(r"\s+#.*$", "", raw_line).strip()
+                if not line or line.startswith("#"):
+                    continue
+                if line.endswith("\\"):
+                    logger.debug("Cannot check multiline requirements in %s.", path)
+                    return False
+                if line.startswith("-"):
+                    try:
+                        parts = shlex.split(line)
+                    except ValueError:
+                        logger.debug("Cannot parse a requirements option in %s.", path)
+                        return False
+                    option = parts[0]
+                    argument = None
+                    if option in ("-r", "--requirement", "-c", "--constraint") and len(parts) == 2:
+                        argument = parts[1]
+                    elif option.startswith(("--requirement=", "--constraint=")) and len(parts) == 1:
+                        option, argument = option.split("=", 1)
+                    elif option.startswith(("-r", "-c")) and len(option) > 2 and len(parts) == 1:
+                        option, argument = option[:2], option[2:]
+                    elif option in _INDEX_FLAGS and len(parts) == 1:
+                        allow_prereleases = allow_prereleases or option == "--pre"
+                        continue
+                    elif (
+                            option in _INDEX_OPTIONS_WITH_VALUE and len(parts) == 2) or (
+                            option.split("=", 1)[0] in _INDEX_OPTIONS_WITH_VALUE
+                            and "=" in option and len(parts) == 1):
+                        continue
+                    if not argument or "://" in argument or "${" in argument:
+                        logger.debug("Cannot inspect a requirements option in %s.", path)
+                        return False
+                    nested_constraint = option in ("-c", "--constraint")
+                    if not read_file(path.parent / argument, nested_constraint):
+                        return False
+                    continue
+                try:
+                    requirement = Requirement(line)
+                except InvalidRequirement:
+                    logger.debug("Cannot parse a requirement in %s.", path)
+                    return False
+                if canonicalize_name(requirement.name) != _PACKAGE:
+                    continue
+                try:
+                    if requirement.marker is not None and not requirement.marker.evaluate():
+                        continue
+                except (UndefinedComparison, UndefinedEnvironmentName, InvalidVersion) as exc:
+                    logger.debug("Cannot evaluate a requirement marker in %s: %s", path, exc)
+                    return False
+                if requirement.url is not None:
+                    logger.debug("Cannot determine the Durable Functions version from a URL in %s.", path)
+                    return False
+                required = required or not constraint
+                specifiers.append(str(requirement.specifier))
+        finally:
+            active.remove(key)
+        visited.add(key)
+        return True
+
+    if not read_file(root / "requirements.txt", False) or not required:
+        return None
+    package_specifiers = SpecifierSet(",".join(specifiers))
+    allow_prereleases = allow_prereleases or any(specifier.prereleases for specifier in package_specifiers)
+    return package_specifiers, allow_prereleases
+
+
+def _allows_later_major(
+        specifiers: SpecifierSet,
+        installed: Version,
+        allow_prereleases: bool = False) -> bool:
+    candidates: set[Version] = set()
+
+    def add_release(epoch: int, major: int, minor: int, patch: int) -> None:
+        prefix = f"{epoch}!" if epoch else ""
+        candidates.add(Version(f"{prefix}{major}.{minor}.{patch}"))
+
+    for major in (installed.major, installed.major + 1, installed.major + 2):
+        add_release(installed.epoch, major, 0, 0)
+        add_release(installed.epoch, major, 0, 1)
+        add_release(installed.epoch, major, 1, 0)
+
+    for specifier in specifiers:
+        try:
+            bound = Version(specifier.version.removesuffix(".*"))
+        except InvalidVersion:
+            continue
+        candidates.add(bound)
+        add_release(bound.epoch, bound.major, bound.minor, bound.micro + 1)
+        add_release(bound.epoch, bound.major, bound.minor + 1, 0)
+        add_release(bound.epoch, bound.major + 1, 0, 0)
+
+    for candidate in tuple(candidates):
+        base = candidate.base_version
+        candidates.update(Version(f"{base}{suffix}") for suffix in (".dev0", "a0", "b0", "rc0"))
+        if candidate.pre and candidate.pre[1] > 0:
+            label, number = candidate.pre
+            candidates.add(Version(f"{base}{label}{number - 1}"))
+
+    allowed = [
+        candidate for candidate in candidates
+        if specifiers.contains(candidate, prereleases=allow_prereleases)
+    ]
+    return (
+        any(candidate.major == installed.major for candidate in allowed)
+        and any(candidate.major > installed.major and candidate > installed for candidate in allowed)
+    )
+
+
+def warn_for_function_directory(function_directory: str, logger: Logger) -> None:
+    """Find the Function App root before checking its dependencies."""
+    roots: list[Path] = []
+    configured_root = os.environ.get("AzureWebJobsScriptRoot")
+    if configured_root:
+        roots.append(Path(configured_root))
+    directory = Path(function_directory)
+    roots.extend((directory, *directory.parents))
+    try:
+        roots.append(Path.cwd())
+    except OSError as exc:
+        logger.debug("Cannot determine the Function App working directory: %s", exc)
+    for root in roots:
+        if (root / "host.json").is_file():
+            warn_if_unbounded(str(root), logger)
+            return
+    logger.debug("Cannot find the Function App root for %s.", function_directory)
+
+
+def warn_if_unbounded(function_directory: str, logger: Logger) -> None:
+    """Warn when the app's requirements demonstrably permit a later major."""
+    root = Path(function_directory)
+    try:
+        requirements = _read_specifiers(root, logger)
+    except RecursionError as exc:
+        logger.debug("Cannot check deeply nested Function App requirements: %s", exc)
+        return
+    if requirements is None:
+        return
+    specifiers, allow_prereleases = requirements
+    try:
+        installed = Version(version(_PACKAGE))
+        allows_later_major = _allows_later_major(specifiers, installed, allow_prereleases)
+    except (PackageNotFoundError, InvalidVersion) as exc:
+        logger.debug("Cannot determine the installed Durable Functions version: %s", exc)
+        return
+    if allows_later_major:
+        next_major = f"{installed.epoch}!" if installed.epoch else ""
+        next_major += str(installed.major + 1)
+        logger.warning(
+            "The Function App requirements allow azure-functions-durable to upgrade beyond major version %d. "
+            "Future major releases may include broad, replay-breaking changes. If a deployment installs these "
+            "requirements without a lock, an automatic upgrade could cause existing orchestrations to fail "
+            "unexpectedly. Pin the major version, for example azure-functions-durable>=%s,<%s, "
+            "or deploy using a complete dependency lock.",
+            installed.major, installed.public, next_major,
+        )

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

@@ -11,16 +11,50 @@
 from concurrent.futures import ThreadPoolExecutor
 from contextvars import ContextVar, copy_context
 from functools import lru_cache, wraps
+from threading import Lock, Thread
 from types import FunctionType
 from typing import Any, Callable, ParamSpec, TypeVar
 
 import azure.functions as func
 
 _invocation_context: ContextVar[func.Context | None] = ContextVar("durable_invocation_context", default=None)
+_dependency_check_lock = Lock()
+_dependency_check_scheduled = False
+_logger = logging.getLogger(__name__)
+# The worker routes SDK logs through the host even after an invocation ends.
+_dependency_logger = logging.getLogger("azure.functions.durable")
 _Parameters = ParamSpec("_Parameters")
 _Result = TypeVar("_Result")
 
 
+def _check_dependency_in_background(directory: str) -> None:
+    from .dependency_warning import warn_for_function_directory
+
+    warn_for_function_directory(directory, _dependency_logger)
+
+
+def _schedule_dependency_check(context: func.Context | None) -> None:
+    global _dependency_check_scheduled
+    if _dependency_check_scheduled or context is None:
+        return
+    directory = getattr(context, "function_directory", None)
+    if not isinstance(directory, str) or not directory:
+        return
+    with _dependency_check_lock:
+        if _dependency_check_scheduled:
+            return
+        _dependency_check_scheduled = True
+        try:
+            Thread(
+                target=_check_dependency_in_background,
+                args=(directory,),
+                name="durable-dependency-check",
+                daemon=True,
+            ).start()
+        except RuntimeError:
+            _logger.warning("Unable to start the Durable Functions dependency check.", exc_info=True)
+
+
 def preserve_function_source(wrapper: Callable[..., Any], original: Callable[..., Any]) -> None:
     """Preserve the filename used by the Functions worker to index the app directory."""
     if isinstance(wrapper, FunctionType) and isinstance(original, FunctionType):
@@ -131,6 +165,7 @@ async def async_wrapper(*args: Any, **kwargs: Any) -> Any:
                 return await function(*original.args, **original.kwargs)
             finally:
                 _invocation_context.reset(token)
+                _schedule_dependency_check(context)
 
         wrapper = async_wrapper
     else:
@@ -142,6 +177,7 @@ def sync_wrapper(*args: Any, **kwargs: Any) -> Any:
                 return function(*original.args, **original.kwargs)
             finally:
                 _invocation_context.reset(token)
+                _schedule_dependency_check(context)
 
         wrapper = sync_wrapper
 

azure-functions-durable/pyproject.toml

@@ -32,7 +32,8 @@ readme = "README.md"
 dependencies = [
     "durabletask[opentelemetry]>=1.13.1",
     "azure-identity>=1.19.0",
-    "azure-functions>=2.4.0b1"
+    "azure-functions>=2.4.0b1",
+    "packaging"
 ]
 
 [project.urls]

tests/azure-functions-durable/e2e/test_logging_e2e.py

@@ -16,6 +16,14 @@ def _assert_host_routed(line: str, message: str) -> None:
     assert re.fullmatch(rf"\[[^\]]+Z\] {re.escape(message)}", line)
 
 
+def test_unpinned_dependency_warning_reaches_the_host(dtask_app):
+    dtask_app.start_orchestration("activity_chain")
+    warning = dtask_app.wait_for_host_log(
+        "The Function App requirements allow azure-functions-durable to upgrade beyond major version")
+    assert "replay-breaking changes" in warning
+    assert "existing orchestrations to fail unexpectedly" in warning
+
+
 def test_client_and_worker_logs_flow_through_host(dtask_app):
     instance_id = dtask_app.start_orchestration("activity_chain")
     status = dtask_app.wait_for_completion(instance_id)