diff --git a/CHANGELOG.md b/CHANGELOG.md index 2e0687e..2e64164 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,16 @@ # Changelog +## 0.2.0 + +- Reject millisecond `receivedAt` values (at or above 100,000,000,000) locally with `InvalidRequestError`. +- Always expose a success `results` dict, including when no elements matched; use safe element access in README examples. +- Retain delivered results for repeat waits for up to one hour, within the existing buffer cap; return copies to protect retained values. +- Pace polls from request completion, rechecking the monotonic gate after every wake, so transport delays cannot shorten the minimum gap. +- Cover HTML, empty, invalid UTF-8 and other non-JSON HTTP error bodies with regression tests; preserve typed errors and status. +- The packaged README includes the correct Mailman → Settings → Tasks API key path. +- `TaskTimeoutError.reason` is `"timeout"` when the server timed the task out and `"deadline"` when the wait ran out, matching the Node and Go SDKs. +- The default `request_timeout` is 15 seconds, matching the guide and the Node and Go SDKs. + ## 0.1.0 - Initial public release of the official Mailman Tasks API client. diff --git a/README.md b/README.md index fc9fc3a..d3f318f 100644 --- a/README.md +++ b/README.md @@ -1,7 +1,7 @@ # Mailman Tasks for Python Official Python 3.8+ client for [Mailman's hosted Tasks API](https://getmailman.app/developers). -Version **0.1.0**, MIT license, zero runtime dependencies. +Version **0.2.0**, MIT license, zero runtime dependencies. ## Install @@ -33,7 +33,7 @@ with MailmanTasks(os.environ["MAILMAN_TASKS_KEY"], group="signup-bot") as client # Trigger the website's email here. result = client.wait(task_id) if result["status"] == "success": - print(result["results"]["code"]) + print(result["results"].get("code")) ``` `wait()` returns any final status. `run(task)` combines create and wait, returning @@ -56,7 +56,7 @@ async def main(): email="alias@example.com", from_="noreply@example.com", subject_includes="verification code", )) - print(result["results"]["code"]) + print(result["results"].get("code")) asyncio.run(main()) ``` @@ -71,12 +71,13 @@ ambiguous: the server may have accepted it. Do not automatically recreate it. ## API and concurrency Both clients accept `MailmanTasks(api_key, *, group="default", -base_url="https://tasks.getmailman.app", request_timeout=30, user_agent=None, +base_url="https://tasks.getmailman.app", request_timeout=15, user_agent=None, opener=None, metadata_limit=50000)`, with async methods on `AsyncMailmanTasks`. - `verify()` returns `True`, or raises, including `AuthError` on 401. - `create_task(task)` returns an id. Missing `receivedAt` is now in Unix **seconds**; missing `timeout` is **300 seconds**; missing `group` uses the client's group. + Values of `receivedAt` at or above 100,000,000,000 are refused as milliseconds. The input is not mutated. Explicit timeout 0 or above 3600 normalizes to 3600, matching the server. Every HTTP request has a **total** `request_timeout` budget covering connection, headers and the complete body, including trickling @@ -120,16 +121,20 @@ independent clients or processes must use different groups to avoid consuming on another's results. Do not change client configuration while it is in use. All poll attempts, including direct `poll()` calls and retries, share the per-group -pacing gate: at least **3 seconds**, or **1 second** immediately after a full page -of 100, and longer when Retry-After/backoff requires it. Backoff and auth delays +pacing gate: at least **3 seconds** after the previous request completes, or +**1 second** immediately after a full page of 100, and longer when Retry-After/backoff +requires it. Backoff and auth delays are shared by direct polls and worker restarts. Polling 401s wait about 10 seconds and keep trying until each wait's deadline. Polling 429s honor Retry-After; 408, 503 and network failures back off 1, 2, 4 ... up to 30 seconds with jitter, subject to the minimum polling interval. -Results enter the bounded dispatcher before waiters are notified. Unclaimed -results live for **one hour**, up to **50,000 per client** across groups; oldest -entries are evicted at capacity. A wait consumes its stored result only after -receiving it successfully, preserving delivery across deadline/cancellation races. +Results enter the bounded dispatcher before waiters are notified. All +results, including delivered results, live for **one hour**, up to **50,000 per client** +across groups; oldest +entries are evicted at capacity. Repeated waits return a copy of the stored result +without another poll, preserving delivery across deadline/cancellation races. +A success always includes a `results` dict; elements that found nothing are absent, +so use `.get("code")` or `.get("loginUrl")` when reading an element. Live task metadata is retained until completion or its original deadline, with `metadata_limit=50000` slots by default (configurable on either client). Concurrent creates reserve slots before sending. At capacity, `create_task` raises a local @@ -140,7 +145,7 @@ ids retain their group and deadline in a separate bounded expiry store (at most on a retained expired id raises `TaskTimeoutError` immediately, even with an explicit longer timeout. After expiry-store eviction/retention ends, an id is unknown and uses the fallback described above. Completed -results retain their group routing in the result buffer until consumed or expired. +results retain their group routing in the result buffer until evicted or expired. Inactive group locks, pacing state and retry counters retire after the pacing gap; one client timer handles retirement without racing workers or queued callers. Consume results promptly; the buffer is not durable storage. @@ -162,7 +167,7 @@ errors), `message`, and `retry_after` in seconds (or `None` when absent). | `NetworkError` | Transport failure, including a client-side request timeout; a create may or may not exist | | `TooManyTasksError` | Client tracking capacity reached; no request was sent | | `TaskFailedError` | Task status `error`; has `error_status`, `error_message`, camelCase aliases `errorStatus`, `errorMessage`, and `result` | -| `TaskTimeoutError` | Server status `timeout`, or local deadline: "Task lost: create it again" | +| `TaskTimeoutError` | `reason` is `"timeout"` for server status `timeout`, or `"deadline"` when the wait ran out: "Task lost: create it again" | Create retries **only explicit 429 or 503 responses**, for at most three HTTP attempts, sleeping for the **longest** of Retry-After and jittered exponential diff --git a/mailman_tasks/__init__.py b/mailman_tasks/__init__.py index 42de491..b3dbeb6 100644 --- a/mailman_tasks/__init__.py +++ b/mailman_tasks/__init__.py @@ -6,7 +6,7 @@ NetworkError, TooManyTasksError, TaskFailedError, TaskTimeoutError, ) -__version__ = "0.1.0" +__version__ = "0.2.0" __all__ = ["MailmanTasks", "AsyncMailmanTasks", "code_task", "link_task", "MailmanError", "AuthError", "InvalidRequestError", "PayloadTooLargeError", "RateLimitError", "PendingLimitError", "ServerBusyError", "RequestTimeoutError", diff --git a/mailman_tasks/client.py b/mailman_tasks/client.py index e9b15f5..57bec1c 100644 --- a/mailman_tasks/client.py +++ b/mailman_tasks/client.py @@ -28,6 +28,13 @@ _NETWORK_ERRORS = (OSError, http.client.HTTPException) +def _copy_result(result: Task) -> Task: + copied = dict(result) + if "results" in result: + copied["results"] = dict(result["results"]) + return copied + + class _NoRedirect(urllib.request.HTTPRedirectHandler): def redirect_request(self, req: Any, fp: Any, code: int, msg: str, headers: Any, newurl: str) -> None: @@ -156,7 +163,7 @@ class MailmanTasks: def __init__(self, api_key: str, *, group: str = "default", base_url: str = "https://tasks.getmailman.app", - request_timeout: float = 30, user_agent: Optional[str] = None, + request_timeout: float = 15, user_agent: Optional[str] = None, opener: Any = None, metadata_limit: int = 50000) -> None: parts = urllib.parse.urlsplit(base_url) if parts.scheme not in ("http", "https") or not parts.netloc or parts.username or parts.password or parts.query or parts.fragment: @@ -177,7 +184,7 @@ def __init__(self, api_key: str, *, group: str = "default", self.group = group self.base_url = base_url.rstrip("/") self.request_timeout = request_timeout - self.user_agent = user_agent or "mailman-tasks-python/0.1.0" + self.user_agent = user_agent or "mailman-tasks-python/0.2.0" self._request_context = threading.local() self._opener = opener or urllib.request.build_opener( _NoRedirect(), _HTTPHandler(self._request_context), _HTTPSHandler(self._request_context)) @@ -416,8 +423,7 @@ def _poll(self, selected: str, *, worker: bool) -> List[Task]: self._expire_waiters(selected) if worker and not self._waiters.get(selected): return [] - started = time.monotonic() - state.next_poll = started + self._poll_interval + state.next_poll = time.monotonic() + self._poll_interval query = urllib.parse.urlencode({"group": selected}) try: results = self._request("GET", "/api/v1/tasks/completed?" + query)["tasks"] @@ -439,12 +445,16 @@ def _poll(self, selected: str, *, worker: bool) -> List[Task]: raise MailmanError("Client is closed; in-flight poll results discarded") state.attempt = 0 interval = self._full_page_interval if len(results) == 100 else self._poll_interval - state.next_poll = started + interval + # Pace from completion so transport dispatch delays cannot + # shorten the gap between actual HTTP attempts. + state.next_poll = time.monotonic() + interval self._prune() for result in results: + if result["status"] == "success": + result.setdefault("results", {}) task_id = result["id"] key = (selected, task_id) - self._buffer[key] = (time.monotonic(), result) + self._buffer[key] = (time.monotonic(), _copy_result(result)) self._buffer.move_to_end(key) self._result_groups[task_id] = selected self._created.pop(task_id, None) @@ -533,20 +543,12 @@ def _wait(self, task_id: str, *, deadline: Optional[float] = None, future, group, duration = self._register(task_id, deadline, timeout, end=end) try: result = future.result(timeout=duration) - return self._consume(task_id, group, result) + return _copy_result(result) except FutureTimeout: raise TaskTimeoutError("Task lost: create it again") from None finally: self._unregister(task_id, group, future) - def _consume(self, task_id: str, group: str, result: Task) -> Task: - with self._condition: - cached = self._buffer.get((group, task_id)) - if cached is not None and cached[1] is result: - self._buffer.pop((group, task_id)) - self._result_groups.pop(task_id, None) - return result - def _group_end(self, group: str) -> float: return min(self._wait_deadlines[f] for waiting in self._waiters[group].values() for f in waiting) @@ -638,7 +640,7 @@ def __exit__(self, *args: Any) -> None: def _outcome(result: Task, redact: Callable[[str], str]) -> Task: if result["status"] == "timeout": - raise TaskTimeoutError("Task timed out: create it again") + raise TaskTimeoutError("Task timed out: no matching email arrived in time", reason="timeout") if result["status"] == "error": safe_result = dict(result) safe_result["results"] = {k: redact(str(v)) for k, v in result.get("results", {}).items()} @@ -649,7 +651,7 @@ def _outcome(result: Task, redact: Callable[[str], str]) -> Task: class AsyncMailmanTasks: """Async facade; HTTP uses an executor, waits do not occupy executor threads.""" def __init__(self, api_key: str, *, group: str = "default", - base_url: str = "https://tasks.getmailman.app", request_timeout: float = 30, + base_url: str = "https://tasks.getmailman.app", request_timeout: float = 15, user_agent: Optional[str] = None, opener: Any = None, metadata_limit: int = 50000) -> None: self._client = MailmanTasks(api_key, group=group, base_url=base_url, @@ -688,13 +690,13 @@ async def _wait(self, task_id: str, *, deadline: Optional[float] = None, future, group, duration = self._client._register(task_id, deadline, timeout, end=end) if future.done(): try: - return self._client._consume(task_id, group, future.result()) + return _copy_result(future.result()) finally: self._client._unregister(task_id, group, future) wrapped = asyncio.wrap_future(future) try: result = await asyncio.wait_for(asyncio.shield(wrapped), duration) - return self._client._consume(task_id, group, result) + return _copy_result(result) except asyncio.TimeoutError: raise TaskTimeoutError("Task lost: create it again") from None finally: diff --git a/mailman_tasks/errors.py b/mailman_tasks/errors.py index 6bfc88c..a5a2893 100644 --- a/mailman_tasks/errors.py +++ b/mailman_tasks/errors.py @@ -57,4 +57,11 @@ def __init__(self, result: Dict[str, Any]) -> None: class TaskTimeoutError(MailmanError): - pass + """No result in time. ``reason`` is ``"timeout"`` when the server completed the + task as timeout (no matching email arrived in time), or ``"deadline"`` when the + wait ended without a result: the task is lost, so trigger a fresh email and + create it again.""" + + def __init__(self, message: str, *, reason: str = "deadline") -> None: + super().__init__(message) + self.reason = reason diff --git a/mailman_tasks/validation.py b/mailman_tasks/validation.py index 485fd96..8c156b9 100644 --- a/mailman_tasks/validation.py +++ b/mailman_tasks/validation.py @@ -40,6 +40,8 @@ def prepare(task: Mapping[str, Any], group: str) -> Dict[str, Any]: invalid(key + " must be an integer in range") if body["receivedAt"] == 0: invalid("receivedAt is required") + if body["receivedAt"] >= 100_000_000_000: + invalid("receivedAt is in seconds, not milliseconds: use int(time.time()) or divide milliseconds by 1000") for key in ("mailFilters", "mailElements"): items = body.get(key, []) if not isinstance(items, list): diff --git a/pyproject.toml b/pyproject.toml index 9f56919..6ab844a 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "mailman-tasks" -version = "0.1.0" +version = "0.2.0" description = "Official Python client for Mailman's hosted Tasks API" readme = "README.md" requires-python = ">=3.8" diff --git a/tests/test_client.py b/tests/test_client.py index 55140c0..65c5239 100644 --- a/tests/test_client.py +++ b/tests/test_client.py @@ -87,7 +87,7 @@ def handle_api(self): self.close_connection = True return code, data, headers = response - encoded = json.dumps(data).encode() + encoded = data if isinstance(data, bytes) else json.dumps(data).encode() try: self.send_response(code) for k, v in headers.items(): @@ -133,7 +133,7 @@ def test_defaults_headers_encoding_and_destructive_poll(self): self.assertEqual(sent[2]['timeout'], 300) self.assertLess(abs(sent[2]['receivedAt'] - time.time()), 2) self.assertEqual(sent[3]['Authorization'], 'Token secret-key') - self.assertEqual(sent[3]['User-Agent'], 'mailman-tasks-python/0.1.0') + self.assertEqual(sent[3]['User-Agent'], 'mailman-tasks-python/0.2.0') self.assertEqual(self.client.poll()[0]['id'], task_id) self.assertEqual(self.client.poll(), []) self.assertIn('group=bot+%26+%2F+%2B', self.state['calls'][-1][1]) @@ -163,6 +163,109 @@ def test_validation_before_transport(self): self.client.create_task(t) self.assertEqual(self.state['calls'][-1][2]['timeout'], 3600) + def test_received_at_milliseconds_guard(self): + for value in (100_000_000_000, int(time.time() * 1000)): + with self.subTest(value=value): + t = task() + t['receivedAt'] = value + with self.assertRaisesRegex(InvalidRequestError, 'seconds, not milliseconds') as caught: + self.client.create_task(t) + self.assertEqual(caught.exception.status, 400) + self.assertEqual(self.count('create'), 0) + t = task() + t['receivedAt'] = 99_999_999_999 + self.client.create_task(t) + self.assertEqual(self.state['calls'][-1][2]['receivedAt'], t['receivedAt']) + + def test_non_json_http_errors_keep_type_status_and_retry_after(self): + for status, error_type in ((400, InvalidRequestError), (413, PayloadTooLargeError), + (500, MailmanError), (502, MailmanError), (503, ServerBusyError)): + for body in (b'secret-key error', b'', b'\xff', b'[1]', b'null'): + with self.subTest(status=status, body=body): + self.state['verify'] = [(status, body, {'Retry-After': '2'})] + with self.assertRaises(error_type) as caught: + self.client.verify() + self.assertIs(type(caught.exception), error_type) + self.assertEqual(caught.exception.status, status) + self.assertEqual(caught.exception.retry_after, 2) + self.assertEqual(str(caught.exception), 'HTTP {}'.format(status)) + + def test_non_json_create_retries_only_explicit_busy_response(self): + for status, error_type in ((400, InvalidRequestError), (413, PayloadTooLargeError)): + before = self.count('create') + self.state['create'] = [(status, b'error', {})] + with self.assertRaises(error_type): + self.client.create_task(task()) + self.assertEqual(self.count('create'), before + 1) + before = self.count('create') + self.state['create'] = [(503, b'busy', {})] + self.client.create_task(task()) + self.assertEqual(self.count('create'), before + 2) + + def test_missing_results_and_repeat_wait_preserve_cached_copy(self): + for status in ('success', 'error', 'timeout'): + with self.subTest(status=status): + t = task() + t['group'] = status + self.state['auto'] = False + task_id = self.client.create_task(t) + self.state['results'][status] = [{'id': task_id, 'status': status}] + polled = self.client.poll(status)[0] + first = self.client.wait(task_id, timeout=1) + if status == 'success': + self.assertEqual(first['results'], {}) + self.assertIsNone(first['results'].get('code')) + polled['results']['code'] = 'changed' + first['results']['code'] = 'changed again' + first['status'] = 'changed' + before = self.count('poll') + second = self.client.wait(task_id, timeout=0) + self.assertEqual(second['status'], status) + if status == 'success': + self.assertEqual(second['results'], {}) + self.assertEqual(self.count('poll'), before) + self.client._buffer_ttl = 0 + with self.assertRaises(TaskTimeoutError): + self.client.wait(task_id, timeout=0) + + def test_run_without_results(self): + self.state['poll'] = [(200, {'tasks': [{'id': '1', 'status': 'success'}]}, {})] + self.assertEqual(self.client.run(task(), timeout=1)['results'], {}) + + def test_poll_gate_rechecks_early_wakes_and_transport_dispatch_delay(self): + clock = [1000.0] + sent = [] + sleeps = [] + self.client._poll_interval = 3 + def request(*args, **kwargs): + clock[0] += .2 # Dispatch can be delayed after the gate opens. + sent.append(clock[0]) + return {'tasks': []} + def early_wake(delay): + sleeps.append(delay) + clock[0] += min(.5, delay) + with patch('mailman_tasks.client.time.monotonic', side_effect=lambda: clock[0]), \ + patch.object(self.client, '_request', side_effect=request), \ + patch.object(self.client, '_retire_groups'), \ + patch.object(self.client._condition, 'wait', side_effect=early_wake): + self.client.poll() + self.client.poll() + self.client.poll() + self.assertGreaterEqual(len(sleeps), 12) + self.assertTrue(all(b - a >= 3 for a, b in zip(sent, sent[1:]))) + + def test_async_missing_results_and_repeat_wait(self): + async def check(): + async with AsyncMailmanTasks('secret-key', base_url=self.client.base_url) as client: + self.state['poll'] = [(200, {'tasks': [{'id': '1', 'status': 'success'}]}, {})] + first = await client.run(task(), timeout=1) + self.assertEqual(first['results'], {}) + first['results']['code'] = 'changed' + before = self.count('poll') + self.assertEqual((await client.wait('1', timeout=0))['results'], {}) + self.assertEqual(self.count('poll'), before) + asyncio.run(check(), debug=True) + def test_error_mapping_and_redaction(self): for code, message, cls in [(400, 'bad secret-key', InvalidRequestError), (413, 'large', PayloadTooLargeError), (429, 'pending task limit exceeded', PendingLimitError), (401, 'bad key', AuthError), @@ -225,8 +328,10 @@ def wait(i): self.assertEqual(self.count('poll'), 1) def test_deadline_and_outcomes(self): - with self.assertRaises(TaskTimeoutError): + self.assertEqual(MailmanTasks('secret-key').request_timeout, 15) + with self.assertRaises(TaskTimeoutError) as lost: self.client.wait('lost', deadline=time.time()+.03) + self.assertEqual(lost.exception.reason, 'deadline') self.state['status'] = 'error' self.state['details'] = {'errorStatus': 'invalid mail task', 'errorMessage': 'bad regex'} with self.assertRaises(TaskFailedError) as caught: @@ -234,8 +339,9 @@ def test_deadline_and_outcomes(self): self.assertEqual(caught.exception.errorStatus, 'invalid mail task') self.assertEqual(caught.exception.error_message, 'bad regex') self.state['status'] = 'timeout' - with self.assertRaises(TaskTimeoutError): + with self.assertRaises(TaskTimeoutError) as timed_out: self.client.run(task(), timeout=1) + self.assertEqual(timed_out.exception.reason, 'timeout') def test_buffer_bound_ttl_and_group_override(self): self.client._buffer_limit = 2 @@ -549,7 +655,7 @@ def open(inner, request, timeout): with self.assertRaises(TaskTimeoutError): self.client.wait('lost', timeout=.02) self.assertEqual(len(requests), 1) - self.assertTrue(29 < requests[0] <= 30, requests) + self.assertTrue(14 < requests[0] <= 15, requests) time.sleep(.03) self.assertEqual(len(requests), 1)