diff --git a/CHANGELOG.md b/CHANGELOG.md index aefb261..14ebdf9 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,55 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ## [Unreleased] +### Added + +- **Deferred-task exception hierarchy** (`jvspatial/exceptions.py`) — + `DeferredTaskError` → `TaskDispatchError` → `TaskSchedulerNotConfiguredError`, + replacing the bare `RuntimeError`s raised by strict dispatch. A strict caller + can now tell "retrying may succeed" (`TaskDispatchError`) from "this + deployment will never dispatch" (`TaskSchedulerNotConfiguredError`). + `DeferredTaskError` also derives from `RuntimeError`, so handlers written + against the previous behavior keep working. + +### Fixed + +- **Strict deferred scheduling only caught the no-op-scheduler case** + (`jvspatial/serverless/`). `dispatch_deferred_task(..., strict=True)` raised + when serverless mode resolved a logging no-op, but every *provider* failure + still logged and returned a synthetic reference — so a caller with its own + failure handling (retry, error signalled upstream, dedup claim released) was + told the task was queued when it had been dropped. `strict` now raises on: + an unset `AWS_LAMBDA_FUNCTION_NAME`; a Lambda `invoke` that raises **or** + answers a non-2xx `StatusCode` / carries a `FunctionError` (an async invoke + returns 202 on acceptance, so boto3 not raising was never proof of + dispatch); an unconfigured SQS client or queue; a failed SQS `send_message`; + a `NoopOrSyncScheduler` with no executor — the scheduler every + *non*-serverless caller gets, which silently dropped strict tasks; and an + EventBridge scheduling failure for a task deferred beyond Lambda's 900s + timeout, where the fallback immediate invoke cannot honor `run_at`. +- **Non-strict dispatch failed differently per transport** + (`jvspatial/serverless/tasks/aws_sqs.py`). SQS `send_message` errors + propagated while the Lambda transport swallowed them, so identical + application code had opposite failure semantics depending on + `JVSPATIAL_AWS_DEFERRED_TRANSPORT`. `strict` is now the single switch on + every transport: `False` is fire-and-forget, `True` raises. +- **The one-time no-op diagnostic never fired for strict callers** + (`jvspatial/serverless/factory.py`). The strict raise preceded + `_note_noop_in_serverless`, so a deployment whose callers are all strict + never got the startup error explaining *why* nothing dispatches. The + diagnostic is now emitted first. + +### Changed + +- **`TaskScheduler.schedule` takes a `strict` argument** + (`jvspatial/serverless/tasks/base.py`). `TaskScheduler` is a public/stable + extension point and `config.task_scheduler` is duck-typed, so + `dispatch_deferred_task` introspects `schedule()` and omits `strict` for + third-party implementations that predate it — those keep serving non-strict + dispatches unchanged. A `strict=True` dispatch through such a scheduler + raises `TaskSchedulerNotConfiguredError` (it cannot honor the guarantee) + rather than `TypeError`. + ## [0.0.11] - 2026-07-02 ### Fixed diff --git a/SPEC.md b/SPEC.md index a20ffcf..73d5c28 100644 --- a/SPEC.md +++ b/SPEC.md @@ -518,6 +518,13 @@ Detection results are memoized via `lru_cache`; tests call `reset_serverless_mod Register handlers with `@deferred_invoke_handler("task.name")`. Handlers **must be idempotent**; the framework provides no exactly-once guarantee. +**Dispatch failure contract** (`strict`, default `False`): + +- **`strict=False`** — fire-and-forget on every transport. A dispatch that cannot be handed to the provider is logged and a synthetic reference returned. +- **`strict=True`** — every such path raises instead. `TaskSchedulerNotConfiguredError` when the deployment can never dispatch (no-op scheduler in serverless mode, `AWS_LAMBDA_FUNCTION_NAME` unset, SQS unconfigured, `NoopOrSyncScheduler` without an executor, or a scheduler predating the `strict` parameter); `TaskDispatchError` when the provider was reached and failed (a raised or non-2xx / `FunctionError` Lambda `invoke`, a raised SQS `send_message`, or an EventBridge failure for a task deferred beyond Lambda's 900s timeout). Both derive from `DeferredTaskError` → `RuntimeError`. + +`TaskScheduler` is a public/stable extension point and `config.task_scheduler` is duck-typed, so `dispatch_deferred_task` introspects `schedule()` and omits `strict` for implementations that predate it. Such a scheduler still serves non-strict dispatches; a `strict=True` dispatch through it raises `TaskSchedulerNotConfiguredError` rather than a `TypeError`. + ### 11.4 Lambda Web Adapter When LWA is detected, `Server` applies best-effort defaults for `AWS_LWA_PASS_THROUGH_PATH` and `AWS_LWA_INVOKE_MODE`. The LWA extension reads these *before* Python starts, so IaC should still set them explicitly for reliability. diff --git a/docs/md/serverless-mode.md b/docs/md/serverless-mode.md index da2b8fc..7d4dd67 100644 --- a/docs/md/serverless-mode.md +++ b/docs/md/serverless-mode.md @@ -74,7 +74,14 @@ Serverless runtimes cannot rely on `asyncio.create_task` for work that must cont - **Not serverless**: the default scheduler is a no-op unless you pass a sync executor to `NoopOrSyncScheduler` or use `override=`. - **AWS**: By default, if `AWS_LAMBDA_FUNCTION_NAME` is set, tasks are sent with **Lambda async invoke** (`InvocationType=Event`). **EventBridge Scheduler** one-shot schedules apply when `run_at` is set, EventBridge is enabled, and role/ARN requirements are satisfied (see below). - **SQS**: Set `JVSPATIAL_AWS_DEFERRED_TRANSPORT=sqs` and `JVSPATIAL_AWS_SQS_QUEUE_URL`; you must run a worker that consumes messages. Messages use a nested `payload` object. Before calling `dispatch_deferred_invoke`, flatten with **`normalize_deferred_envelope`** (exported from `jvspatial`) so the body matches the Lambda/LWA shape. -- **Strict dispatch**: `create_task("…", {}, strict=True)` (or `dispatch_deferred_task(..., strict=True)`) raises `RuntimeError` if serverless mode is on but the resolved scheduler is a logging no-op. Otherwise the first no-op schedule in serverless mode emits a one-time **error** log. +- **Strict dispatch**: `create_task("…", {}, strict=True)` (or `dispatch_deferred_task(..., strict=True)`) raises whenever the task could not be handed to the provider, instead of returning a synthetic reference. Pass it when you have your own failure handling — retrying, signalling an error upstream, releasing a dedup claim — and a dropped task would be data loss. Without it, dispatch stays fire-and-forget on every transport and the first no-op schedule in serverless mode emits a one-time **error** log. + + | Exception | When | + |---|---| + | `TaskSchedulerNotConfiguredError` | The deployment can never dispatch: a logging no-op resolved in serverless mode, `AWS_LAMBDA_FUNCTION_NAME` unset, SQS client/queue unconfigured, a `NoopOrSyncScheduler` with no executor, or a scheduler whose `schedule()` predates the `strict` parameter. Retrying will not help. | + | `TaskDispatchError` | The provider was reached and the call failed: a Lambda `invoke` that raised or answered non-2xx / `FunctionError`, an SQS `send_message` that raised, or an EventBridge schedule that failed for a task deferred past Lambda's 900s timeout (an immediate invoke cannot honor `run_at`). | + + Both derive from `DeferredTaskError`, which derives from `RuntimeError` — handlers written against the earlier bare-`RuntimeError` behavior keep working. Third-party `TaskScheduler` implementations written before `strict` existed are detected by signature and called without it, so non-strict dispatch is unaffected. - **Provider override**: `JVSPATIAL_DEFERRED_TASK_PROVIDER` (`aws`, `azure`, `gcp`, `vercel`, `auto`) or `Config.deferred_task_provider` / `ServerConfig.deferred_task_provider`. - **Detection**: `detect_serverless_provider()` complements `is_serverless_mode()` for choosing a backend. diff --git a/jvspatial/exceptions.py b/jvspatial/exceptions.py index acd8d89..a269c0c 100644 --- a/jvspatial/exceptions.py +++ b/jvspatial/exceptions.py @@ -418,6 +418,47 @@ def __init__(self, config_key: str, details: Optional[Dict[str, Any]] = None): self.config_key = config_key +# ============================================================================= +# SERVERLESS / DEFERRED TASK EXCEPTIONS +# ============================================================================= + + +class DeferredTaskError(JVSpatialError, RuntimeError): + """Base exception for deferred task dispatch. + + Also inherits :class:`RuntimeError` so existing ``except RuntimeError`` + handlers around ``dispatch_deferred_task`` keep working — the strict + no-op guard raised a bare ``RuntimeError`` before this hierarchy existed. + """ + + +class TaskDispatchError(DeferredTaskError): + """Raised when a task was handed to the provider and the provider failed. + + Distinguishable from :class:`TaskSchedulerNotConfiguredError` so a strict + caller can tell "retry may succeed" from "this deployment will never + dispatch". Only raised when the caller passed ``strict=True``; the + default fire-and-forget contract logs and returns a synthetic reference. + """ + + def __init__( + self, task_type: str, reason: str, details: Optional[Dict[str, Any]] = None + ): + message = f"Deferred task {task_type!r} was not dispatched: {reason}" + super().__init__(message, details) + self.task_type = task_type + self.reason = reason + + +class TaskSchedulerNotConfiguredError(TaskDispatchError): + """Raised when the resolved scheduler cannot dispatch at all. + + A missing ``AWS_LAMBDA_FUNCTION_NAME``, an unconfigured SQS client or + queue URL, or a logging no-op scheduler resolved while serverless mode is + on. Retrying will not help; the deployment needs configuration. + """ + + # ============================================================================= # EXPORTS # ============================================================================= @@ -472,4 +513,8 @@ def __init__(self, config_key: str, details: Optional[Dict[str, Any]] = None): # Configuration exceptions "InvalidConfigurationError", "MissingConfigurationError", + # Serverless / deferred task exceptions + "DeferredTaskError", + "TaskDispatchError", + "TaskSchedulerNotConfiguredError", ] diff --git a/jvspatial/serverless/factory.py b/jvspatial/serverless/factory.py index 4da32ba..dade0be 100644 --- a/jvspatial/serverless/factory.py +++ b/jvspatial/serverless/factory.py @@ -2,10 +2,13 @@ from __future__ import annotations +import inspect import logging -from typing import Any, Optional +from functools import lru_cache +from typing import Any, Dict, Optional from jvspatial.env import env +from jvspatial.exceptions import TaskSchedulerNotConfiguredError from jvspatial.runtime.serverless import detect_serverless_provider, is_serverless_mode from .tasks.aws_lambda import AwsLambdaDeferredTaskScheduler @@ -111,6 +114,31 @@ def get_task_scheduler( ) +@lru_cache(maxsize=None) +def _scheduler_accepts_strict(sched_type: type) -> bool: + """Whether ``sched_type.schedule`` takes a ``strict`` argument. + + ``TaskScheduler`` is public/stable (``docs/md/stability.md``) and + ``config.task_scheduler`` is duck-typed, so third-party schedulers written + against the pre-``strict`` signature are in the wild. Passing ``strict=`` + unconditionally would break every one of them on every dispatch, including + non-strict ones. Cached per class — this is on the dispatch path. + + Unintrospectable callables (C extensions, some mocks) are assumed modern: + the ABC declares the parameter, so that is the better default. + """ + schedule = getattr(sched_type, "schedule", None) + if schedule is None: + return True + try: + params = inspect.signature(schedule).parameters + except (TypeError, ValueError): # pragma: no cover - exotic callables + return True + if any(p.kind is inspect.Parameter.VAR_KEYWORD for p in params.values()): + return True + return "strict" in params + + def _note_noop_in_serverless(sched: TaskScheduler, config: Optional[Any]) -> None: global _NOOP_DEFERRED_LOGGED if not isinstance(sched, LoggingNoopTaskScheduler): @@ -140,25 +168,54 @@ def dispatch_deferred_task( """Schedule a JSON-serializable deferred task; thin wrapper over :func:`get_task_scheduler`. Args: - strict: If True and serverless mode is on but the resolved scheduler is - :class:`LoggingNoopTaskScheduler`, raise ``RuntimeError`` instead of - returning a synthetic reference. + strict: If True, scheduling failures raise instead of returning a + synthetic reference: the noop-scheduler-in-serverless case (as + before), AND provider dispatch failures — a missing + ``AWS_LAMBDA_FUNCTION_NAME``, a failed or rejected Lambda + ``invoke``, an unconfigured SQS client, a failed + ``send_message``, a non-serverless ``NoopOrSyncScheduler`` with no + executor. Callers passing ``strict=True`` have their own failure + handling; a synthetic reference for an undispatched task turns + that handling into silent data loss. + + Raises: + TaskSchedulerNotConfiguredError: ``strict`` and the deployment cannot + dispatch at all — including when the resolved scheduler predates + the ``strict`` parameter and so cannot honor the guarantee. + TaskDispatchError: ``strict`` and the provider rejected or failed the + call. Both derive from ``RuntimeError``, so handlers written + against the previous bare-``RuntimeError`` raise still work. """ sched = get_task_scheduler(config, override=override) + # Emit the once-per-process diagnostic before the strict raise, so a + # deployment whose callers are all strict still gets the startup error + # telling it *why* nothing dispatches. + _note_noop_in_serverless(sched, config) if ( strict and is_serverless_mode(config) and isinstance(sched, LoggingNoopTaskScheduler) ): - raise RuntimeError( - "Deferred task scheduler is a no-op in serverless mode; configure an AWS " - "transport (Lambda/SQS), or inject config.task_scheduler." + raise TaskSchedulerNotConfiguredError( + task_type, + "the resolved scheduler is a no-op in serverless mode; configure an " + "AWS transport (Lambda/SQS), or inject config.task_scheduler", ) - _note_noop_in_serverless(sched, config) - return sched.schedule( - task_type, - payload, - delay_seconds=delay_seconds, - retry_config=retry_config, - run_at=run_at, - ) + + kwargs: Dict[str, Any] = { + "delay_seconds": delay_seconds, + "retry_config": retry_config, + "run_at": run_at, + } + if _scheduler_accepts_strict(type(sched)): + kwargs["strict"] = strict + elif strict: + # A scheduler predating the ``strict`` parameter cannot honor the + # guarantee the caller is asking for. Say so, rather than passing an + # argument it will reject with TypeError. + raise TaskSchedulerNotConfiguredError( + task_type, + f"{type(sched).__name__}.schedule() does not accept 'strict'; it " + "predates strict scheduling and cannot guarantee dispatch", + ) + return sched.schedule(task_type, payload, **kwargs) diff --git a/jvspatial/serverless/tasks/aws_lambda.py b/jvspatial/serverless/tasks/aws_lambda.py index 964f810..65029f6 100644 --- a/jvspatial/serverless/tasks/aws_lambda.py +++ b/jvspatial/serverless/tasks/aws_lambda.py @@ -8,16 +8,45 @@ from typing import Any, Dict, Optional from jvspatial.env import env, parse_bool, resolve_aws_region +from jvspatial.exceptions import TaskDispatchError, TaskSchedulerNotConfiguredError from jvspatial.runtime.eventbridge_readiness import resolve_eventbridge_lambda_arn from .base import RetryConfig, TaskScheduler logger = logging.getLogger(__name__) +# Hard ceiling on a single Lambda execution. A task deferred further out than +# this cannot be honored by invoking now and waiting inside the handler. +_LAMBDA_MAX_TIMEOUT_SECONDS = 900 + _lambda_client_cache: list[Optional[Any]] = [None] _scheduler_client_cache: list[Optional[Any]] = [None] +def _invoke_rejection(response: Any) -> Optional[str]: + """Describe why an async ``invoke`` response is a failure, else ``None``. + + ``InvocationType="Event"`` returns ``202`` when Lambda has accepted the + invocation. Anything else — or a ``FunctionError`` — means the task was + not queued, even though boto3 did not raise. + """ + if not isinstance(response, dict): + return None + function_error = response.get("FunctionError") + if function_error: + return f"Lambda returned FunctionError={function_error!r}" + status = response.get("StatusCode") + if status is None: + return None + try: + status_int = int(status) + except (TypeError, ValueError): + return f"Lambda returned a non-numeric StatusCode={status!r}" + if 200 <= status_int < 300: + return None + return f"Lambda returned StatusCode={status_int}" + + def _get_lambda_client() -> Any: if _lambda_client_cache[0] is None: import boto3 @@ -91,9 +120,9 @@ def _create_eventbridge_schedule( bridge_input: Dict[str, Any] = {**payload, "task_type": task_type} else: bridge_input = {"task_type": task_type, "payload": payload} - # Match Lambda async-invoke body: handlers (e.g. WhatsApp media_batch) use - # process_at to avoid sleeping media_batch_window again after EventBridge - # already fired at run_at. + # Match the Lambda async-invoke body: a handler reads process_at to avoid + # re-waiting its own batching window after EventBridge already fired at + # run_at. bridge_input["process_at"] = run_at try: @@ -151,13 +180,26 @@ def schedule( delay_seconds: int = 0, retry_config: Optional[RetryConfig] = None, run_at: Optional[float] = None, + strict: bool = False, ) -> str: - """Dispatch via Lambda async invoke or EventBridge Scheduler; see base class.""" + """Dispatch via Lambda async invoke or EventBridge Scheduler; see base class. + + Under ``strict=True`` every path that fails to hand the task to AWS + raises instead of returning a synthetic reference. A caller that opted + into strict has failure handling of its own — signalling an error back + to an upstream sender, releasing a dedup claim so a retry is accepted + — and a reference for a task that was never dispatched converts that + handling into silent data loss. + """ reference = f"aws-lambda-{uuid.uuid4()}" if retry_config is not None: pass # reserved for future retry metadata on envelope if not self._function_name: + if strict: + raise TaskSchedulerNotConfiguredError( + task_type, "AWS_LAMBDA_FUNCTION_NAME is not set" + ) logger.warning( "AWS_LAMBDA_FUNCTION_NAME not set; deferred task %s not dispatched", task_type, @@ -168,19 +210,32 @@ def schedule( if effective_run_at is None and delay_seconds > 0: effective_run_at = time.time() + delay_seconds - if effective_run_at is not None and _create_eventbridge_schedule( - task_type, payload, effective_run_at, reference - ): - return reference + if effective_run_at is not None: + if _create_eventbridge_schedule( + task_type, payload, effective_run_at, reference + ): + return reference + # EventBridge failed, so we fall back to invoking now with + # ``process_at`` in the body and let the handler wait. That only + # works inside a single Lambda execution: past the maximum + # timeout the handler cannot survive until ``run_at``, so the + # task is doomed and a strict caller must hear about it. Shorter + # delays fall through to the invoke below, which strict guards. + if strict and effective_run_at - time.time() > _LAMBDA_MAX_TIMEOUT_SECONDS: + raise TaskDispatchError( + task_type, + "EventBridge scheduling failed and the requested delay " + f"exceeds the {_LAMBDA_MAX_TIMEOUT_SECONDS}s Lambda " + "timeout, so an immediate invoke cannot honor run_at", + ) body = _build_invoke_body(task_type, payload, effective_run_at) try: - self._client().invoke( + response = self._client().invoke( FunctionName=self._function_name, InvocationType="Event", Payload=json.dumps(body), ) - logger.info("Invoked deferred task %s (ref=%s)", task_type, reference) except Exception as e: logger.error( "Failed Lambda invoke for deferred task %s: %s", @@ -188,4 +243,21 @@ def schedule( e, exc_info=True, ) + if strict: + raise TaskDispatchError( + task_type, f"Lambda invoke raised {type(e).__name__}: {e}" + ) from e + return reference + + # A raised exception is not the only failure mode: an async invoke + # answers 202 on acceptance, and a rejected or errored invocation + # comes back as a non-2xx StatusCode or a FunctionError field. + rejection = _invoke_rejection(response) + if rejection is not None: + logger.error("Lambda rejected deferred task %s: %s", task_type, rejection) + if strict: + raise TaskDispatchError(task_type, rejection) + return reference + + logger.info("Invoked deferred task %s (ref=%s)", task_type, reference) return reference diff --git a/jvspatial/serverless/tasks/aws_sqs.py b/jvspatial/serverless/tasks/aws_sqs.py index b2cb17b..58d3739 100644 --- a/jvspatial/serverless/tasks/aws_sqs.py +++ b/jvspatial/serverless/tasks/aws_sqs.py @@ -1,12 +1,17 @@ """AWS SQS-backed deferred task scheduler.""" import json +import logging import time import uuid from typing import Any, Optional +from jvspatial.exceptions import TaskDispatchError, TaskSchedulerNotConfiguredError + from .base import RetryConfig, TaskScheduler +logger = logging.getLogger(__name__) + # SQS maximum per-message delay _SQS_MAX_DELAY_SECONDS = 900 @@ -28,10 +33,22 @@ def schedule( delay_seconds: int = 0, retry_config: Optional[RetryConfig] = None, run_at: Optional[float] = None, + strict: bool = False, ) -> str: - """Enqueue a message on SQS with optional delay; see base class.""" + """Enqueue a message on SQS with optional delay; see base class. + + ``strict`` is the single switch that decides whether a failed dispatch + raises, on every transport. Previously ``send_message`` failures + propagated here while the Lambda transport swallowed them, so the same + application code had opposite failure semantics depending on + ``JVSPATIAL_AWS_DEFERRED_TRANSPORT``. + """ reference = f"aws-sqs-{uuid.uuid4()}" if not self._sqs_client or not self._queue_url: + if strict: + raise TaskSchedulerNotConfiguredError( + task_type, "SQS client or queue URL is not configured" + ) return reference delay = max(0, int(delay_seconds)) @@ -46,9 +63,21 @@ def schedule( "reference": reference, "run_at": run_at, } - self._sqs_client.send_message( - QueueUrl=self._queue_url, - MessageBody=json.dumps(message), - DelaySeconds=delay, - ) + try: + self._sqs_client.send_message( + QueueUrl=self._queue_url, + MessageBody=json.dumps(message), + DelaySeconds=delay, + ) + except Exception as e: + logger.error( + "Failed SQS send_message for deferred task %s: %s", + task_type, + e, + exc_info=True, + ) + if strict: + raise TaskDispatchError( + task_type, f"SQS send_message raised {type(e).__name__}: {e}" + ) from e return reference diff --git a/jvspatial/serverless/tasks/base.py b/jvspatial/serverless/tasks/base.py index 94bd723..7c3f0a0 100644 --- a/jvspatial/serverless/tasks/base.py +++ b/jvspatial/serverless/tasks/base.py @@ -27,14 +27,32 @@ def schedule( delay_seconds: int = 0, retry_config: Optional[RetryConfig] = None, run_at: Optional[float] = None, + strict: bool = False, ) -> str: """Schedule a task and return provider reference id. Args: - task_type: Stable namespaced task id (e.g. ``app.whatsapp.media_batch``). + task_type: Stable namespaced task id (e.g. ``app.media.batch``). payload: JSON-serializable task input. delay_seconds: Minimum delay before execution (relative), when ``run_at`` unset. retry_config: Optional retry metadata for queue-based backends. run_at: Optional Unix epoch seconds for absolute execution time; backends map this to native scheduling (e.g. EventBridge) or embed in the message. + strict: When True, a dispatch that cannot be handed to the provider + MUST raise instead of logging and returning a synthetic + reference. A caller passing ``strict=True`` is stating that it + has fallback behaviour of its own — retrying, signalling an + error upstream, releasing a dedup claim — and that a + silently-dropped task is data loss. The default preserves + fire-and-forget semantics. + + Raise :class:`~jvspatial.exceptions.TaskSchedulerNotConfiguredError` + when the deployment can never dispatch (no retry will help) and + :class:`~jvspatial.exceptions.TaskDispatchError` when the + provider was reached and rejected or failed the call. + + Implementations added after this parameter existed should keep + ``strict`` in the signature. ``dispatch_deferred_task`` introspects the + signature and omits the argument for schedulers that predate it, so + third-party implementations keep working for non-strict dispatches. """ diff --git a/jvspatial/serverless/tasks/stub.py b/jvspatial/serverless/tasks/stub.py index 7087c96..e50eea6 100644 --- a/jvspatial/serverless/tasks/stub.py +++ b/jvspatial/serverless/tasks/stub.py @@ -4,6 +4,8 @@ import uuid from typing import Any, Optional +from jvspatial.exceptions import TaskSchedulerNotConfiguredError + from .base import RetryConfig, TaskScheduler logger = logging.getLogger(__name__) @@ -22,6 +24,7 @@ def schedule( delay_seconds: int = 0, retry_config: Optional[RetryConfig] = None, run_at: Optional[float] = None, + strict: bool = False, ) -> str: """Log and return a synthetic reference; see base class. @@ -30,6 +33,13 @@ def schedule( once-per-process startup error from ``serverless.factory._note_noop_in_serverless`` is sufficient (audit §7.14 / SPEC §11.2). + + ``dispatch_deferred_task`` guards the same condition earlier and with + more context, so this raise is the backstop for direct callers and + for a no-op injected via ``config.task_scheduler`` outside serverless + mode — where the factory's ``is_serverless_mode`` gate does not fire. """ + if strict: + raise TaskSchedulerNotConfiguredError(task_type, self._message) logger.debug("%s (task_type=%s)", self._message, task_type) return f"noop-{uuid.uuid4()}" diff --git a/jvspatial/serverless/tasks/sync.py b/jvspatial/serverless/tasks/sync.py index aa5a66a..8051bd6 100644 --- a/jvspatial/serverless/tasks/sync.py +++ b/jvspatial/serverless/tasks/sync.py @@ -3,11 +3,21 @@ import uuid from typing import Any, Callable, Optional +from jvspatial.exceptions import TaskSchedulerNotConfiguredError + from .base import RetryConfig, TaskScheduler class NoopOrSyncScheduler(TaskScheduler): - """Fallback scheduler that executes handlers inline.""" + """Fallback scheduler that executes handlers inline, or drops them. + + With an ``executor`` this runs the task in-process, which satisfies + ``strict``: the work happened before ``schedule`` returned. Without one + — the shape :func:`~jvspatial.serverless.factory.get_task_scheduler` + returns for every non-serverless caller — nothing runs at all, so a + ``strict`` dispatch raises rather than handing back a reference for work + that will never happen. + """ def __init__(self, executor: Optional[Callable[[str, Any], Any]] = None): self._executor = executor @@ -19,10 +29,19 @@ def schedule( delay_seconds: int = 0, retry_config: Optional[RetryConfig] = None, run_at: Optional[float] = None, + strict: bool = False, ) -> str: """Run the configured executor immediately; see base class.""" reference = f"sync-{uuid.uuid4()}" - if self._executor is not None: - # Strict-safe default: execute immediately in-process. - self._executor(task_type, payload) + if self._executor is None: + if strict: + raise TaskSchedulerNotConfiguredError( + task_type, + "no executor is configured on NoopOrSyncScheduler, so the " + "task would be silently dropped; inject " + "config.task_scheduler or enable serverless mode", + ) + return reference + # Executed in-process, so the strict guarantee is already met. + self._executor(task_type, payload) return reference diff --git a/tests/serverless/test_aws_lambda_scheduler.py b/tests/serverless/test_aws_lambda_scheduler.py index 333eadb..2b037a0 100644 --- a/tests/serverless/test_aws_lambda_scheduler.py +++ b/tests/serverless/test_aws_lambda_scheduler.py @@ -7,9 +7,18 @@ import pytest +from jvspatial.exceptions import TaskDispatchError, TaskSchedulerNotConfiguredError from jvspatial.runtime.serverless import reset_serverless_mode_cache from jvspatial.serverless.tasks.aws_lambda import AwsLambdaDeferredTaskScheduler +try: # botocore ships with boto3; fall back so the suite runs without it. + from botocore.exceptions import ClientError +except ImportError: # pragma: no cover - exercised only without boto3 + + class ClientError(Exception): # type: ignore[no-redef] + def __init__(self, error_response, operation_name): + super().__init__(f"{operation_name}: {error_response}") + @pytest.fixture(autouse=True) def _clear_serverless_caches(): @@ -18,12 +27,6 @@ def _clear_serverless_caches(): reset_serverless_mode_cache() -def test_schedule_invoke_without_function_name_logs(): - sched = AwsLambdaDeferredTaskScheduler(function_name="") - ref = sched.schedule("t.example", {"a": 1}) - assert ref.startswith("aws-lambda-") - - def test_schedule_lambda_invoke_payload_merges_dict(): mock_client = MagicMock() sched = AwsLambdaDeferredTaskScheduler( @@ -31,8 +34,8 @@ def test_schedule_lambda_invoke_payload_merges_dict(): lambda_client=mock_client, ) sched.schedule( - "app.whatsapp.media_batch", - {"sender": "u1", "media_batch_window": 1.5}, + "app.media.batch", + {"sender": "u1", "batch_window": 1.5}, run_at=12345.0, ) mock_client.invoke.assert_called_once() @@ -40,9 +43,9 @@ def test_schedule_lambda_invoke_payload_merges_dict(): assert call_kw["FunctionName"] == "fn" assert call_kw["InvocationType"] == "Event" body = json.loads(call_kw["Payload"]) - assert body["task_type"] == "app.whatsapp.media_batch" + assert body["task_type"] == "app.media.batch" assert body["sender"] == "u1" - assert body["media_batch_window"] == 1.5 + assert body["batch_window"] == 1.5 assert body["process_at"] == 12345.0 @@ -92,3 +95,139 @@ def test_delay_seconds_becomes_process_at_in_payload(): sched.schedule("t", {"x": 1}, delay_seconds=30) body = json.loads(mock_client.invoke.call_args.kwargs["Payload"]) assert body["process_at"] == 1030.0 + + +# ── strict semantics ──────────────────────────────────────────────────────── +# +# strict=True is the caller stating that it has its own failure handling — +# signalling an error back to an upstream sender, releasing a dedup claim so a +# retry is accepted — and that an undispatched task must therefore RAISE +# rather than hand back a synthetic reference. Returning a reference for work +# that was never dispatched turns that handling into silent data loss. + + +def test_strict_raises_when_function_name_unset(monkeypatch): + monkeypatch.delenv("AWS_LAMBDA_FUNCTION_NAME", raising=False) + sched = AwsLambdaDeferredTaskScheduler(function_name="") + with pytest.raises( + TaskSchedulerNotConfiguredError, match="AWS_LAMBDA_FUNCTION_NAME" + ): + sched.schedule("t.task", {"k": "v"}, strict=True) + + +def test_non_strict_keeps_fire_and_forget_when_function_name_unset(monkeypatch): + monkeypatch.delenv("AWS_LAMBDA_FUNCTION_NAME", raising=False) + sched = AwsLambdaDeferredTaskScheduler(function_name="") + ref = sched.schedule("t.task", {"k": "v"}) + assert ref.startswith("aws-lambda-") + + +def test_strict_raises_task_dispatch_error_on_invoke_failure(): + client = MagicMock() + client.invoke.side_effect = ClientError( + {"Error": {"Code": "ServiceException", "Message": "boom"}}, "Invoke" + ) + sched = AwsLambdaDeferredTaskScheduler(function_name="fn", lambda_client=client) + with pytest.raises(TaskDispatchError) as excinfo: + sched.schedule("t.task", {"k": "v"}, strict=True) + assert excinfo.value.task_type == "t.task" + # Still a RuntimeError, so pre-existing handlers keep working. + assert isinstance(excinfo.value, RuntimeError) + + +def test_non_strict_swallows_invoke_failure_unchanged(): + client = MagicMock() + client.invoke.side_effect = ClientError( + {"Error": {"Code": "ServiceException", "Message": "boom"}}, "Invoke" + ) + sched = AwsLambdaDeferredTaskScheduler(function_name="fn", lambda_client=client) + ref = sched.schedule("t.task", {"k": "v"}) + assert ref.startswith("aws-lambda-") + + +def test_strict_success_returns_reference(): + client = MagicMock() + client.invoke.return_value = {"StatusCode": 202} + sched = AwsLambdaDeferredTaskScheduler(function_name="fn", lambda_client=client) + ref = sched.schedule("t.task", {"k": "v"}, strict=True) + assert ref.startswith("aws-lambda-") + client.invoke.assert_called_once() + + +# ── invoke responses that do not raise but are still failures ─────────────── + + +@pytest.mark.parametrize( + "response", + [ + {"StatusCode": 500}, + {"StatusCode": 202, "FunctionError": "Unhandled"}, + {"StatusCode": "not-a-number"}, + ], +) +def test_strict_raises_on_rejected_invoke_response(response): + """boto3 returns rather than raising when Lambda rejects the invocation.""" + client = MagicMock() + client.invoke.return_value = response + sched = AwsLambdaDeferredTaskScheduler(function_name="fn", lambda_client=client) + with pytest.raises(TaskDispatchError): + sched.schedule("t.task", {"k": "v"}, strict=True) + + +@pytest.mark.parametrize("response", [{"StatusCode": 200}, {"StatusCode": 202}]) +def test_accepted_invoke_response_is_not_a_failure(response): + client = MagicMock() + client.invoke.return_value = response + sched = AwsLambdaDeferredTaskScheduler(function_name="fn", lambda_client=client) + assert sched.schedule("t.task", {"k": "v"}, strict=True).startswith("aws-lambda-") + + +def test_non_strict_ignores_rejected_invoke_response(): + client = MagicMock() + client.invoke.return_value = {"StatusCode": 500} + sched = AwsLambdaDeferredTaskScheduler(function_name="fn", lambda_client=client) + assert sched.schedule("t.task", {"k": "v"}).startswith("aws-lambda-") + + +# ── EventBridge failure must not silently become a doomed immediate invoke ─── + + +def test_strict_raises_when_eventbridge_fails_beyond_lambda_timeout(): + """A far-future task cannot be honored by invoking now and waiting.""" + client = MagicMock() + client.invoke.return_value = {"StatusCode": 202} + sched = AwsLambdaDeferredTaskScheduler(function_name="fn", lambda_client=client) + with patch( + "jvspatial.serverless.tasks.aws_lambda._create_eventbridge_schedule", + return_value=False, + ): + with pytest.raises(TaskDispatchError, match="EventBridge"): + sched.schedule("t.task", {"k": "v"}, delay_seconds=3600, strict=True) + client.invoke.assert_not_called() + + +def test_strict_allows_short_delay_fallback_to_immediate_invoke(): + """Inside the Lambda timeout the handler can honor process_at itself.""" + client = MagicMock() + client.invoke.return_value = {"StatusCode": 202} + sched = AwsLambdaDeferredTaskScheduler(function_name="fn", lambda_client=client) + with patch( + "jvspatial.serverless.tasks.aws_lambda._create_eventbridge_schedule", + return_value=False, + ): + ref = sched.schedule("t.task", {"k": "v"}, delay_seconds=30, strict=True) + assert ref.startswith("aws-lambda-") + client.invoke.assert_called_once() + + +def test_non_strict_still_falls_back_for_a_far_future_task(): + client = MagicMock() + client.invoke.return_value = {"StatusCode": 202} + sched = AwsLambdaDeferredTaskScheduler(function_name="fn", lambda_client=client) + with patch( + "jvspatial.serverless.tasks.aws_lambda._create_eventbridge_schedule", + return_value=False, + ): + ref = sched.schedule("t.task", {"k": "v"}, delay_seconds=3600) + assert ref.startswith("aws-lambda-") + client.invoke.assert_called_once() diff --git a/tests/serverless/test_task_scheduler_strict.py b/tests/serverless/test_task_scheduler_strict.py new file mode 100644 index 0000000..e5c95e4 --- /dev/null +++ b/tests/serverless/test_task_scheduler_strict.py @@ -0,0 +1,210 @@ +"""Strict dispatch semantics across every scheduler and through the factory. + +``strict=True`` is the caller stating it has its own failure handling and that +a silently-dropped task is data loss. That guarantee is only worth anything if +it holds on *every* transport and survives the factory, so these cases cover +the adapters the Lambda-specific suite does not, plus the +``dispatch_deferred_task`` plumbing itself. +""" + +from typing import Any, Optional +from unittest.mock import MagicMock + +import pytest + +from jvspatial.exceptions import ( + DeferredTaskError, + TaskDispatchError, + TaskSchedulerNotConfiguredError, +) +from jvspatial.runtime.serverless import reset_serverless_mode_cache +from jvspatial.serverless.factory import dispatch_deferred_task +from jvspatial.serverless.tasks.aws_sqs import AwsSqsTaskScheduler +from jvspatial.serverless.tasks.base import RetryConfig, TaskScheduler +from jvspatial.serverless.tasks.stub import LoggingNoopTaskScheduler +from jvspatial.serverless.tasks.sync import NoopOrSyncScheduler + + +@pytest.fixture(autouse=True) +def _clear_serverless_caches(): + reset_serverless_mode_cache() + yield + reset_serverless_mode_cache() + + +# ── SQS ───────────────────────────────────────────────────────────────────── + + +def test_sqs_strict_raises_when_client_unconfigured(): + sched = AwsSqsTaskScheduler(sqs_client=None, queue_url=None) + with pytest.raises(TaskSchedulerNotConfiguredError, match="SQS"): + sched.schedule("t.task", {"k": "v"}, strict=True) + + +def test_sqs_non_strict_returns_reference_when_unconfigured(): + sched = AwsSqsTaskScheduler(sqs_client=None, queue_url=None) + assert sched.schedule("t.task", {"k": "v"}).startswith("aws-sqs-") + + +def test_sqs_strict_raises_on_send_message_failure(): + client = MagicMock() + client.send_message.side_effect = RuntimeError("throttled") + sched = AwsSqsTaskScheduler(sqs_client=client, queue_url="https://q") + with pytest.raises(TaskDispatchError, match="send_message"): + sched.schedule("t.task", {"k": "v"}, strict=True) + + +def test_sqs_non_strict_swallows_send_message_failure(): + """Fire-and-forget is now the non-strict contract on every transport. + + Previously SQS propagated while the Lambda transport swallowed, so the + same application code had opposite semantics depending on + ``JVSPATIAL_AWS_DEFERRED_TRANSPORT``. + """ + client = MagicMock() + client.send_message.side_effect = RuntimeError("throttled") + sched = AwsSqsTaskScheduler(sqs_client=client, queue_url="https://q") + assert sched.schedule("t.task", {"k": "v"}).startswith("aws-sqs-") + + +def test_sqs_success_sends_and_returns_reference(): + client = MagicMock() + sched = AwsSqsTaskScheduler(sqs_client=client, queue_url="https://q") + ref = sched.schedule("t.task", {"k": "v"}, strict=True) + assert ref.startswith("aws-sqs-") + client.send_message.assert_called_once() + + +# ── logging no-op ─────────────────────────────────────────────────────────── + + +def test_stub_strict_raises(): + sched = LoggingNoopTaskScheduler("no transport configured") + with pytest.raises(TaskSchedulerNotConfiguredError, match="no transport"): + sched.schedule("t.task", {"k": "v"}, strict=True) + + +def test_stub_non_strict_returns_reference(): + sched = LoggingNoopTaskScheduler() + assert sched.schedule("t.task", {"k": "v"}).startswith("noop-") + + +# ── sync / no-op fallback ─────────────────────────────────────────────────── + + +def test_sync_without_executor_raises_under_strict(): + """The shape every non-serverless caller gets: nothing would run.""" + sched = NoopOrSyncScheduler(None) + with pytest.raises(TaskSchedulerNotConfiguredError, match="no executor"): + sched.schedule("t.task", {"k": "v"}, strict=True) + + +def test_sync_without_executor_is_still_fire_and_forget_by_default(): + sched = NoopOrSyncScheduler(None) + assert sched.schedule("t.task", {"k": "v"}).startswith("sync-") + + +def test_sync_with_executor_satisfies_strict(): + """The work happened in-process before schedule() returned.""" + seen = [] + sched = NoopOrSyncScheduler(lambda t, p: seen.append((t, p))) + ref = sched.schedule("t.task", {"k": "v"}, strict=True) + assert ref.startswith("sync-") + assert seen == [("t.task", {"k": "v"})] + + +# ── factory plumbing ──────────────────────────────────────────────────────── + + +class _RecordingScheduler(TaskScheduler): + """Current-signature scheduler that records what it was handed.""" + + def __init__(self) -> None: + self.calls: list = [] + + def schedule( + self, + task_type: str, + payload: Any, + delay_seconds: int = 0, + retry_config: Optional[RetryConfig] = None, + run_at: Optional[float] = None, + strict: bool = False, + ) -> str: + self.calls.append({"task_type": task_type, "strict": strict}) + return "recorded" + + +class _LegacyScheduler: + """Third-party scheduler written against the pre-``strict`` signature.""" + + def __init__(self) -> None: + self.calls: list = [] + + def schedule( + self, + task_type: str, + payload: Any, + delay_seconds: int = 0, + retry_config: Optional[RetryConfig] = None, + run_at: Optional[float] = None, + ) -> str: + self.calls.append(task_type) + return "legacy" + + +class _KwargsScheduler: + """Scheduler that absorbs unknown keywords.""" + + def __init__(self) -> None: + self.calls: list = [] + + def schedule(self, task_type: str, payload: Any, **kwargs: Any) -> str: + self.calls.append(kwargs) + return "kwargs" + + +@pytest.mark.parametrize("strict", [False, True]) +def test_dispatch_forwards_strict_to_a_modern_scheduler(strict): + sched = _RecordingScheduler() + assert ( + dispatch_deferred_task("t.task", {"k": "v"}, override=sched, strict=strict) + == "recorded" + ) + assert sched.calls == [{"task_type": "t.task", "strict": strict}] + + +def test_dispatch_omits_strict_for_a_legacy_scheduler(): + """A pre-``strict`` third-party scheduler must keep working. + + ``TaskScheduler`` is public/stable and ``config.task_scheduler`` is + duck-typed, so forwarding ``strict=`` unconditionally would raise + ``TypeError`` on every dispatch — including non-strict ones. + """ + sched = _LegacyScheduler() + assert dispatch_deferred_task("t.task", {"k": "v"}, override=sched) == "legacy" + assert sched.calls == ["t.task"] + + +def test_dispatch_refuses_strict_on_a_legacy_scheduler(): + """It cannot honor the guarantee, so say so instead of pretending.""" + sched = _LegacyScheduler() + with pytest.raises(TaskSchedulerNotConfiguredError, match="does not accept"): + dispatch_deferred_task("t.task", {"k": "v"}, override=sched, strict=True) + assert sched.calls == [] + + +def test_dispatch_forwards_strict_to_a_kwargs_scheduler(): + sched = _KwargsScheduler() + assert ( + dispatch_deferred_task("t.task", {"k": "v"}, override=sched, strict=True) + == "kwargs" + ) + assert sched.calls[0]["strict"] is True + + +def test_deferred_task_errors_are_runtime_errors(): + """Handlers written against the previous bare-RuntimeError raise still work.""" + assert issubclass(DeferredTaskError, RuntimeError) + assert issubclass(TaskDispatchError, DeferredTaskError) + assert issubclass(TaskSchedulerNotConfiguredError, TaskDispatchError)