diff --git a/agent_memory_leaderboard/README.md b/agent_memory_leaderboard/README.md index b8d549a..6d16837 100644 --- a/agent_memory_leaderboard/README.md +++ b/agent_memory_leaderboard/README.md @@ -164,10 +164,33 @@ the `idempotent_v1` protocol in SQLite **before** sending HTTP. On a later Add retry, an attempt without an acknowledged memory ID can safely resend the same key and body: the backend returns the existing receipt or creates the record once. This recovers both pre-commit server failures and a crash after journaling -but before sending. The adapter still makes one write attempt per incoming Add -attempt; it does not add an internal retry loop. Once the memory ID is known, +but before sending. Ordinary errors still end the current Add attempt; only the +specific catalog-visibility errors described below are retried internally. +Once the memory ID is known, retries use point reads and retrieval probes instead of reposting. +When Memoria returns HTTP 500 with a recognized MatrixOne missing `mem_*` table +error (1064/42000 with the missing-table parser message, or 1146/42S02 with +`no such table`), the adapter uses a bounded recovery window configured by +`MEMORIA_CATALOG_RETRY_TIMEOUT_SECONDS` (default 5 seconds, finite in `[0, 120]`; +0 disables retries). Helm exposes the same value as `catalogRetryTimeoutSeconds`. +The window starts on the first recognized error and is shared by every backend +call and chunk in one Add/Search request; it never resets per retry or chunk. +Backoff starts at 0.5 seconds and is capped at 1 second and the remaining window. +Retry admission waits and HTTP I/O limits are capped by that same deadline. +As with the request budget, synchronous I/O may unwind after expiry; a late +success is rejected, and the window is not a hard cancellation timer. +This covers atomic keyed writes, point reads, +recovery queries and Search; a write retry preserves its exact payload and +idempotency key. Backoff releases the backend slot and consumes the existing +request budget and, during verification, the original visibility budget. Other +server errors, SQL syntax errors, authorization/validation failures and +non-atomic writes do not enter this retry loop. Exhaustion keeps the existing +error mapping and ambiguous-write journal state; it never claims the write was +not committed. Retry warnings contain only attempt numbers and delay durations. +This is bounded recovery for transient visibility, not a guarantee that a +permanently missing schema will recover or a replacement for backend consistency. + Atomic writes, all point reads/recovery queries/visibility probes, and public Search explicitly use branch `main`, keeping the receipt scope independent of a backend checkout. diff --git a/agent_memory_leaderboard/src/budget.py b/agent_memory_leaderboard/src/budget.py index 38d125d..f55c835 100644 --- a/agent_memory_leaderboard/src/budget.py +++ b/agent_memory_leaderboard/src/budget.py @@ -7,12 +7,40 @@ import time from contextlib import contextmanager from contextvars import ContextVar +from dataclasses import dataclass from .errors import AdapterError DEADLINE: ContextVar[float | None] = ContextVar("adapter_request_deadline", default=None) +@dataclass +class CatalogRetryWindow: + seconds: float + deadline: float | None = None + + def start(self) -> float: + if self.deadline is None: + self.deadline = time.monotonic() + self.seconds + return self.deadline + + +CATALOG_RETRY_WINDOW: ContextVar[CatalogRetryWindow | None] = ContextVar( + "adapter_catalog_retry_window", default=None +) + + +@contextmanager +def catalog_retry_budget(seconds: float): + """One lazy window shared by all backend calls/chunks in this request.""" + parent = CATALOG_RETRY_WINDOW.get() + token = CATALOG_RETRY_WINDOW.set(parent if parent is not None else CatalogRetryWindow(seconds)) + try: + yield + finally: + CATALOG_RETRY_WINDOW.reset(token) + + class RequestBudgetExceeded(AdapterError): def __init__(self): super().__init__( diff --git a/agent_memory_leaderboard/src/config.py b/agent_memory_leaderboard/src/config.py index b4b5b6b..2917df5 100644 --- a/agent_memory_leaderboard/src/config.py +++ b/agent_memory_leaderboard/src/config.py @@ -16,6 +16,7 @@ class Settings: connect_timeout_seconds: float = field(default=30, kw_only=True) request_timeout_seconds: float = field(default=1700, kw_only=True) visibility_timeout_seconds: float = field(default=60, kw_only=True) + catalog_retry_timeout_seconds: float = field(default=5, kw_only=True) backend_max_concurrency: int = 0 backend_requests_per_second: float = 0 backend_queue_timeout_seconds: float = 120 @@ -58,6 +59,11 @@ def __post_init__(self) -> None: raise ValueError("ADAPTER_JOURNAL_LOG_ALL_OPERATIONS must be a boolean") if not math.isfinite(self.journal_slow_ms) or self.journal_slow_ms < 0: raise ValueError("ADAPTER_JOURNAL_SLOW_MS must be nonnegative and finite") + if ( + not math.isfinite(self.catalog_retry_timeout_seconds) + or not 0 <= self.catalog_retry_timeout_seconds <= 120 + ): + raise ValueError("MEMORIA_CATALOG_RETRY_TIMEOUT_SECONDS must be in [0, 120]") if ( not math.isfinite(self.visibility_timeout_seconds) or not 0 < self.visibility_timeout_seconds <= 120 @@ -89,6 +95,9 @@ def from_env(cls) -> "Settings": connect_timeout_seconds=float(os.getenv("MEMORIA_CONNECT_TIMEOUT_SECONDS", "30")), request_timeout_seconds=float(os.getenv("ADAPTER_REQUEST_TIMEOUT_SECONDS", "1700")), visibility_timeout_seconds=float(os.getenv("MEMORIA_VISIBILITY_TIMEOUT_SECONDS", "60")), + catalog_retry_timeout_seconds=float( + os.getenv("MEMORIA_CATALOG_RETRY_TIMEOUT_SECONDS", "5") + ), backend_max_concurrency=int(os.getenv("MEMORIA_MAX_CONCURRENCY", "0")), backend_requests_per_second=float(os.getenv("MEMORIA_REQUESTS_PER_SECOND", "0")), backend_queue_timeout_seconds=float(os.getenv("MEMORIA_QUEUE_TIMEOUT_SECONDS", "120")), diff --git a/agent_memory_leaderboard/src/errors.py b/agent_memory_leaderboard/src/errors.py index 54aa5e0..66e1ea7 100644 --- a/agent_memory_leaderboard/src/errors.py +++ b/agent_memory_leaderboard/src/errors.py @@ -17,3 +17,7 @@ def __init__( class VisibilityPending(AdapterError): """A valid empty read/retrieval result; other 503s must not enter this retry loop.""" + + +class CatalogVisibilityPending(AdapterError): + """Recognized upstream missing-table error; a write may already have committed.""" diff --git a/agent_memory_leaderboard/src/memoria.py b/agent_memory_leaderboard/src/memoria.py index 4203148..22f8206 100644 --- a/agent_memory_leaderboard/src/memoria.py +++ b/agent_memory_leaderboard/src/memoria.py @@ -1,6 +1,8 @@ """HTTP boundary: use an owner-scoped master credential, never caller headers.""" +import logging import math +import re import time from typing import Annotated, Any from urllib.parse import quote @@ -8,14 +10,23 @@ import httpx from pydantic import BaseModel, ConfigDict, Field, ValidationError +from .budget import CATALOG_RETRY_WINDOW, CatalogRetryWindow from .budget import remaining as budget_remaining from .config import Settings -from .errors import AdapterError, VisibilityPending +from .errors import AdapterError, CatalogVisibilityPending, VisibilityPending from .models import Nonblank from .scheduling import BackendScheduler IDEMPOTENT_PATH = "/v1/memories/idempotent" ADAPTER_BRANCH = "main" +_TABLE_IDENTIFIER = r'''["'`]*(?:[\w-]+["'`]*\.)?["'`]*mem_[a-z0-9_]+["'`]*''' +_MISSING_TABLE = re.compile( + rf"\b1064\s*\(42000\):\s*SQL parser error:\s*table\s+{_TABLE_IDENTIFIER}" + rf"\s+does not exist\b|\b1146\s*\(42S02\):\s*no such table\s+" + rf"{_TABLE_IDENTIFIER}(?=\s|[,;:)]|$)", + re.IGNORECASE, +) +logger = logging.getLogger(__name__) class Memory(BaseModel): @@ -58,14 +69,88 @@ def request( body: dict | None = None, *, deadline: float | None = None, + ) -> Any: + # Replay only reads and the atomic write route, preserving the exact + # payload/key. A failed response does not prove a write was uncommitted. + safe_replay = method == "GET" or ( + method == "POST" + and ( + path in {"/v1/memories/query", "/v1/memories/search"} + or (path == IDEMPOTENT_PATH and body and body.get("idempotency_key")) + ) + ) + window = CATALOG_RETRY_WINDOW.get() + if window is None: + window = CatalogRetryWindow(self.settings.catalog_retry_timeout_seconds) + retry_deadline = None + attempt = 0 + while True: + try: + result = self._request_once( + method, path, owner, body, deadline=deadline, retry_deadline=retry_deadline + ) + if retry_deadline is not None and time.monotonic() >= retry_deadline: + budget_remaining(float("inf")) + if deadline is not None: + self._remaining_visibility(deadline) + raise CatalogVisibilityPending(502, "Memoria catalog retry window expired") + return result + except CatalogVisibilityPending: + if not safe_replay: + raise + retry_deadline = window.start() + available = budget_remaining(retry_deadline - time.monotonic()) + if deadline is not None: + available = min(available, self._remaining_visibility(deadline)) + if available <= 0: + raise + delay = min(0.5 if attempt == 0 else 1.0, available) + # No upstream body, URL, owner, credentials or payload in logs. + logger.warning( + "Memoria catalog visibility retry wait: attempt=%d delay_ms=%.0f", + attempt + 1, + delay * 1000, + ) + # _request_once released the backend slot before this wait. + time.sleep(delay) + attempt += 1 + budget_remaining(float("inf")) + if deadline is not None: + self._remaining_visibility(deadline) + if time.monotonic() >= retry_deadline: + raise + except AdapterError as exc: + # Earlier attempts were ambiguous even if this retry never sent. + if retry_deadline is not None: + exc.write_not_sent = False + raise + + def _request_once( + self, + method: str, + path: str, + owner: str, + body: dict | None = None, + *, + deadline: float | None = None, + retry_deadline: float | None = None, ) -> Any: budget_remaining(float("inf")) if deadline is not None: self._remaining_visibility(deadline) visibility_limited = False try: - with self.scheduler.slot(deadline=deadline): + admission_deadline = min( + deadline if deadline is not None else math.inf, + retry_deadline if retry_deadline is not None else math.inf, + ) + with self.scheduler.slot(deadline=admission_deadline): timeout = budget_remaining(self.settings.timeout_seconds) + if retry_deadline is not None: + retry_remaining = retry_deadline - time.monotonic() + if retry_remaining <= 0: + raise CatalogVisibilityPending(502, "Memoria catalog retry window expired") + timeout = min(timeout, retry_remaining) if deadline is not None: remaining = self._remaining_visibility(deadline) visibility_limited = remaining <= timeout @@ -126,7 +211,12 @@ def request( # conflicts and server errors remain ambiguous and require recovery. rejected = code in {400, 401, 403, 404, 405, 413, 415, 422, 425, 429} retry_after = response.headers.get("Retry-After", "") - raise AdapterError( + error_class = ( + CatalogVisibilityPending + if code == 500 and _MISSING_TABLE.search(response.text[:16384]) + else AdapterError + ) + raise error_class( status, reason, write_not_sent=rejected, diff --git a/agent_memory_leaderboard/src/service.py b/agent_memory_leaderboard/src/service.py index f508ebd..b5b423e 100644 --- a/agent_memory_leaderboard/src/service.py +++ b/agent_memory_leaderboard/src/service.py @@ -2,8 +2,8 @@ import json from itertools import chain, pairwise +from .budget import catalog_retry_budget, request_budget from .budget import remaining as budget_remaining -from .budget import request_budget from .config import Settings from .errors import AdapterError from .ingestion import message_parts @@ -30,7 +30,11 @@ def owner(self, user_id: str) -> str: return digest([self.settings.namespace, user_id]) def add(self, request: AddRequest) -> AddResponse: - with request_budget(self.settings.request_timeout_seconds), self.journal.metrics.add(): + with ( + request_budget(self.settings.request_timeout_seconds), + catalog_retry_budget(self.settings.catalog_retry_timeout_seconds), + self.journal.metrics.add(), + ): owner = self.owner(request.user_id) key = digest([self.settings.memoria_url.rstrip("/"), owner, request.request_id]) fingerprint = digest(request.model_dump()) @@ -125,7 +129,10 @@ def write( ) def search(self, request: SearchRequest) -> SearchResponse: - with request_budget(self.settings.request_timeout_seconds): + with ( + request_budget(self.settings.request_timeout_seconds), + catalog_retry_budget(self.settings.catalog_retry_timeout_seconds), + ): result = self._search(request) budget_remaining(float("inf")) return result diff --git a/agent_memory_leaderboard/tests/test_catalog_retry.py b/agent_memory_leaderboard/tests/test_catalog_retry.py new file mode 100644 index 0000000..ef5216b --- /dev/null +++ b/agent_memory_leaderboard/tests/test_catalog_retry.py @@ -0,0 +1,348 @@ +"""Synthetic catalog-lag faults; no environment or evaluation payloads.""" + +from dataclasses import replace + +import httpx +import pytest + +from src import budget as budget_module +from src import memoria as memoria_module +from src import scheduling as scheduling_module +from src.budget import request_budget +from src.errors import AdapterError +from src.journal import IDEMPOTENT_PROTOCOL, Journal +from src.memoria import IDEMPOTENT_PATH, Memoria +from src.models import AddRequest +from src.service import digest +from tests.test_endpoints import addition, running +from tests.test_endpoints import settings as settings +from tests.test_endpoints import upstream as upstream +from tests.test_write_visibility import Clock + +PARSER_MISSING = ( + 'Database error: error returned from database: 1064 (42000): ' + 'SQL parser error: table "mem_memories" does not exist' +) +EXECUTOR_MISSING = ( + "Database error: error returned from database: 1146 (42S02): " + "no such table synthetic_db.mem_memories" +) + + +@pytest.fixture +def clock(monkeypatch): + value = Clock() + monkeypatch.setattr(memoria_module, "time", value) + monkeypatch.setattr(budget_module, "time", value) + monkeypatch.setattr(scheduling_module, "time", value) + return value + + +@pytest.mark.parametrize("error", [PARSER_MISSING, EXECUTOR_MISSING]) +@pytest.mark.parametrize("committed", [False, True]) +def test_atomic_write_retries_identical_payload_without_duplicates( + settings, upstream, clock, error, committed, caplog +): + def fail_once(request, body, owner): + if request.url.path == IDEMPOTENT_PATH and len(upstream.writes) == 1: + if committed: + upstream.respond(request, body, owner) + return httpx.Response(500, text=error + " synthetic-sensitive-diagnostic") + + upstream.hook = fail_once + with running(settings, upstream) as client: + assert client.post("/add", json=addition()).status_code == 200 + assert client.post("/add", json=addition()).status_code == 200 + assert clock.sleeps == [0.5] + assert len(upstream.writes) == 2 + assert upstream.writes[0] == upstream.writes[1] + assert len(upstream.memories) == len(upstream.receipts) == 1 + assert "catalog visibility retry" in caplog.text + assert "synthetic-sensitive-diagnostic" not in caplog.text + assert "synthetic_db" not in caplog.text + assert settings.memoria_master_key not in caplog.text + assert upstream.writes[0][3]["idempotency_key"] not in caplog.text + + +@pytest.mark.parametrize("stage", ["read", "search"]) +def test_add_visibility_missing_table_retries_without_reposting( + settings, upstream, clock, stage +): + attempts = [] + + def lag(request, body, owner): + selected = request.method == "GET" if stage == "read" else ( + request.url.path == "/v1/memories/search" + ) + if selected: + attempts.append((str(request.url), body)) + if len(attempts) == 1: + return httpx.Response(500, text=PARSER_MISSING) + + upstream.hook = lag + with running(settings, upstream) as client: + assert client.post("/add", json=addition()).status_code == 200 + assert len(attempts) == 2 and attempts[0] == attempts[1] + assert len(upstream.writes) == len(upstream.memories) == 1 + assert clock.sleeps == [0.5] + + +def test_search_endpoint_recovers_catalog_error(settings, upstream, clock): + with running(settings, upstream) as client: + assert client.post("/add", json=addition()).status_code == 200 + calls = [] + + def lag(request, body, owner): + if request.url.path == "/v1/memories/search": + calls.append(body) + if len(calls) == 1: + return httpx.Response(500, text=EXECUTOR_MISSING) + + upstream.hook = lag + response = client.post( + "/search", json={"user_id": "alice", "query": "moon", "top_k": 100} + ) + assert response.status_code == 200 and len(response.json()["data"]) == 1 + assert len(calls) == 2 and calls[0] == calls[1] + assert len(upstream.writes) == 1 + + +def test_retry_exhaustion_preserves_atomic_intent_and_later_replay( + settings, upstream, clock +): + def unavailable(request, body, owner): + if request.url.path == IDEMPOTENT_PATH: + upstream.respond(request, body, owner) + return httpx.Response(500, text=EXECUTOR_MISSING) + + upstream.hook = unavailable + model = AddRequest.model_validate(addition()) + with running(settings, upstream) as client: + response = client.post("/add", json=addition()) + assert response.status_code == 502 + assert "synthetic_db" not in response.text + assert "mem_memories" not in response.text + owner = digest([settings.namespace, model.user_id]) + key = digest([settings.memoria_url, owner, model.request_id]) + assert Journal(settings.state_dir).state(key, 0) == ( + "inflight", None, IDEMPOTENT_PROTOCOL + ) + assert len(upstream.writes) == 6 + assert sum(clock.sleeps) == 5 + assert clock.now == 105 + upstream.hook = None + assert client.post("/add", json=addition()).status_code == 200 + assert len(upstream.writes) == 7 and len(upstream.memories) == 1 + assert all(call == upstream.writes[0] for call in upstream.writes) + + +@pytest.mark.parametrize( + "status,error", + [ + (500, "unrelated internal failure"), + (500, '1064 (42000): SQL parser error: syntax error near "mem_memories"'), + (500, '1064 (42000): SQL parser error: table "unrelated_table" does not exist'), + (500, "1146 (42S02): no such table synthetic_db.mem_memories_suffix.not_a_table"), + (401, PARSER_MISSING), + (404, PARSER_MISSING), + (422, PARSER_MISSING), + (429, PARSER_MISSING), + (503, PARSER_MISSING), + ], +) +def test_unrelated_errors_are_not_retried(settings, upstream, clock, status, error): + upstream.hook = lambda *_: httpx.Response(status, text=error) + with running(settings, upstream) as client: + assert client.post("/add", json=addition()).status_code != 200 + assert len(upstream.writes) == 1 and not clock.sleeps + + +@pytest.mark.parametrize( + "path,body", [("/v1/memories", {}), (IDEMPOTENT_PATH, {"content": "synthetic"})] +) +def test_non_atomic_or_unkeyed_write_is_never_retried(settings, clock, path, body): + calls = [] + + def fail(request): + calls.append(request) + return httpx.Response(500, text=PARSER_MISSING) + + with httpx.Client(transport=httpx.MockTransport(fail)) as client: + with pytest.raises(AdapterError) as caught: + Memoria(settings, client).request("POST", path, "a" * 64, body) + assert caught.value.write_not_sent is False + assert len(calls) == 1 and not clock.sleeps + + +@pytest.mark.parametrize("limited_by", ["request", "visibility"]) +def test_retry_obeys_existing_deadlines_and_releases_scheduler_slot( + settings, clock, limited_by +): + calls = [] + + def fail(request): + calls.append(request) + return httpx.Response(500, text=EXECUTOR_MISSING) + + with httpx.Client(transport=httpx.MockTransport(fail)) as client: + boundary = Memoria(replace(settings, backend_max_concurrency=1), client) + sleep = clock.sleep + + def check_sleep(seconds): + assert boundary.scheduler.active == 0 + sleep(seconds) + + clock.sleep = check_sleep + with request_budget(0.25 if limited_by == "request" else 10): + with pytest.raises(AdapterError) as caught: + boundary.request( + "GET", "/v1/memories/synthetic", "a" * 64, + deadline=clock.now + 0.25 if limited_by == "visibility" else None, + ) + assert caught.value.status == (504 if limited_by == "request" else 503) + assert len(calls) == 1 and clock.sleeps == [0.25] + assert boundary.scheduler.active == 0 + + +def test_all_chunks_and_verification_share_one_request_window(settings, upstream, clock): + write_calls = [] + + def lag(request, body, owner): + if request.url.path == IDEMPOTENT_PATH: + write_calls.append(body) + if len(write_calls) == 1: + return httpx.Response(500, text=PARSER_MISSING) + if request.method == "GET": + return httpx.Response(500, text=EXECUTOR_MISSING) + + upstream.hook = lag + with running(settings, upstream) as client: + assert client.post("/add", json=addition()).status_code == 502 + assert clock.now == 105 and sum(clock.sleeps) == 5 + assert len(upstream.writes) == 2 and len(upstream.memories) == 1 + upstream.hook = None + assert client.post("/add", json=addition()).status_code == 200 + assert len(upstream.writes) == 2 + + +def test_multi_message_add_does_not_reset_retry_window_per_chunk(settings, upstream, clock): + def lag(request, body, owner): + if request.url.path == IDEMPOTENT_PATH: + if len(upstream.writes) == 1 or body["extra_metadata"]["message_index"] == 1: + return httpx.Response(500, text=PARSER_MISSING) + + upstream.hook = lag + payload = addition(messages=[{"role": "user", "content": f"fact {i}"} for i in range(2)]) + with running(settings, upstream) as client: + assert client.post("/add", json=payload).status_code == 502 + assert clock.now == 105 and sum(clock.sleeps) == 5 + assert len(upstream.memories) == 1 + + +@pytest.mark.parametrize("seconds", [0, 0.2, 2, 5]) +def test_configured_window_bounds_entire_recovery(settings, clock, seconds): + calls = [] + + def lag(request): + calls.append(request) + return httpx.Response(500, text=PARSER_MISSING) + + with httpx.Client(transport=httpx.MockTransport(lag)) as client: + boundary = Memoria(replace(settings, catalog_retry_timeout_seconds=seconds), client) + with pytest.raises(AdapterError): + boundary.request("GET", "/v1/memories/synthetic", "a" * 64) + assert clock.now == pytest.approx(100 + seconds) + assert sum(clock.sleeps) == pytest.approx(seconds) + if seconds == 0: + assert len(calls) == 1 and not clock.sleeps + for request in calls[1:]: + assert max(request.extensions["timeout"].values()) <= seconds + + +def test_retry_http_time_consumes_window_and_late_success_is_not_accepted(settings, clock): + calls = [] + + def lag(request): + calls.append(request) + if len(calls) == 1: + return httpx.Response(500, text=PARSER_MISSING) + assert max(request.extensions["timeout"].values()) == 4.5 + clock.now += 4.6 + return httpx.Response(200, json={"synthetic": True}) + + with httpx.Client(transport=httpx.MockTransport(lag)) as client: + with pytest.raises(AdapterError, match="catalog retry window expired"): + Memoria(settings, client).request("GET", "/v1/memories/synthetic", "a" * 64) + assert len(calls) == 2 and clock.sleeps == [0.5] + + +def test_retry_http_failures_consume_the_same_window(settings, clock): + calls = [] + + def lag(request): + calls.append(request) + if len(calls) > 1: + clock.now += min(1.0, max(request.extensions["timeout"].values())) + return httpx.Response(500, text=PARSER_MISSING) + + with httpx.Client(transport=httpx.MockTransport(lag)) as client: + with pytest.raises(AdapterError): + Memoria(settings, client).request("GET", "/v1/memories/synthetic", "a" * 64) + assert clock.now == 105 + assert sum(clock.sleeps) < 5 + + +def test_retry_connect_failure_keeps_prior_write_outcome_ambiguous(settings, clock): + calls = [] + + def fail(request): + calls.append(request) + if len(calls) == 1: + return httpx.Response(500, text=PARSER_MISSING) + raise httpx.ConnectError("synthetic connection failure", request=request) + + with httpx.Client(transport=httpx.MockTransport(fail)) as client: + with pytest.raises(AdapterError) as caught: + Memoria(settings, client).store_idempotent( + "a" * 64, {"content": "synthetic"}, "synthetic-key" + ) + assert caught.value.status == 503 and caught.value.write_not_sent is False + assert len(calls) == 2 + + +def test_later_incoming_request_gets_a_fresh_window(settings, upstream, clock): + def unavailable(request, body, owner): + if request.url.path == IDEMPOTENT_PATH: + return httpx.Response(500, text=PARSER_MISSING) + + upstream.hook = unavailable + with running(settings, upstream) as client: + assert client.post("/add", json=addition()).status_code == 502 + assert clock.now == 105 + failures = [] + + def once(request, body, owner): + if request.url.path == IDEMPOTENT_PATH: + failures.append(True) + if len(failures) == 1: + return httpx.Response(500, text=PARSER_MISSING) + + upstream.hook = once + assert client.post("/add", json=addition()).status_code == 200 + assert clock.now == 105.5 and len(upstream.memories) == 1 + + +@pytest.mark.parametrize("value", [-1, 121, float("nan"), float("inf")]) +def test_invalid_catalog_retry_window_rejected(settings, value): + with pytest.raises(ValueError, match="MEMORIA_CATALOG_RETRY_TIMEOUT_SECONDS"): + replace(settings, catalog_retry_timeout_seconds=value) + + +def test_catalog_retry_window_from_environment(monkeypatch): + from src.config import Settings + + monkeypatch.setenv("MEMORIA_URL", "http://memoria.test") + monkeypatch.setenv("MEMORIA_MASTER_KEY", "synthetic") + assert Settings.from_env().catalog_retry_timeout_seconds == 5 + monkeypatch.setenv("MEMORIA_CATALOG_RETRY_TIMEOUT_SECONDS", "2.5") + assert Settings.from_env().catalog_retry_timeout_seconds == 2.5 diff --git a/deploy/chart/templates/deployment.yaml b/deploy/chart/templates/deployment.yaml index d565b5a..6d96d07 100644 --- a/deploy/chart/templates/deployment.yaml +++ b/deploy/chart/templates/deployment.yaml @@ -76,6 +76,8 @@ spec: value: {{ .Values.namespacePrefix | quote }} - name: MEMORIA_TIMEOUT_SECONDS value: {{ .Values.timeoutSeconds | quote }} + - name: MEMORIA_CATALOG_RETRY_TIMEOUT_SECONDS + value: {{ .Values.catalogRetryTimeoutSeconds | quote }} volumeMounts: - name: state mountPath: /app/var diff --git a/deploy/chart/values.schema.json b/deploy/chart/values.schema.json index 6a56562..2d521fa 100644 --- a/deploy/chart/values.schema.json +++ b/deploy/chart/values.schema.json @@ -2,6 +2,7 @@ "$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "properties": { + "catalogRetryTimeoutSeconds": {"type": "number", "minimum": 0, "maximum": 120}, "workers": {"type": "integer", "minimum": 1} }, "required": ["workers"] diff --git a/deploy/chart/values.yaml b/deploy/chart/values.yaml index 93ee15d..bbcf13e 100644 --- a/deploy/chart/values.yaml +++ b/deploy/chart/values.yaml @@ -34,6 +34,7 @@ certificate: issuer: aliyun-prod-letsencrypt namespacePrefix: aml-dev-v1 timeoutSeconds: 120 +catalogRetryTimeoutSeconds: 5 workers: 1 resources: requests: