diff --git a/packages/client/README.md b/packages/client/README.md index ca834fe..75311a2 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -538,10 +538,21 @@ objects through the `SkillStore` interface and cannot tell which store produced customer-confidential. A mobile key (`mob-…`) or a client-side environment ID raises from the constructor. -**The SDK key goes only where you pointed it.** `base_uri` must be `https://` (plain `http://` -is refused, except to a loopback host for a local test double), and redirects are never -followed, so a 3xx from a proxy or a misconfigured private instance stops delivery rather than -forwarding the key to whatever host the `Location` header names. +**The SDK key goes only where you pointed it.** `base_uri` and `stream_uri` must each be +`https://` (plain `http://` is refused, except to a loopback host for a local test double), +and redirects are never followed, so a 3xx from a proxy or a misconfigured private instance +stops delivery rather than forwarding the key to whatever host the `Location` header names. + +**Polling and streaming have separate hosts.** LaunchDarkly serves `/sdk/poll` from +`https://sdk.launchdarkly.com` and `/sdk/stream` from `https://stream.launchdarkly.com`, so +the defaults are a pair. Pass `base_uri` on its own and it applies to both — what a relay or a +private instance serving both endpoints from one host needs — or pass `stream_uri` as well to +override them independently. + +**`close()` is final.** A closed store still answers from the content it received, but +delivery cannot be resumed: `start()` afterwards raises. That is what gives `close()` a +postcondition you can rely on — delivery has stopped — even when its join times out. +Construct a new store to resume. **Reads are memory-bounded.** No poll body or streamed event is held past `MAX_RESPONSE_BYTES` (64 MiB, far above any real payload); one that crosses it is dropped without being applied, the @@ -582,16 +593,16 @@ Windows. | Export | Description | |---|---| -| `skill_refs(config)` | Project a config's `skills` array into `list[SkillReference]`. Pure — no client, store, or network needed. Returns `[]` when absent. | +| `skill_refs(config)` | Project a config's `skills` array into `list[SkillReference]`. Pure — no client, store, or network needed. Returns `[]` when the field is absent. A `skills` field that is present but not an array — including an explicit `null` — fails the config parse instead, so a field the SDK could not read never reaches a pruning reconcile as "no skills". | | `get_skill(key, *, version=None)` | One verified skill, or `None`. `version=None` means newest available; a specific `version` matches exactly. Raises only when no store is configured. | | `get_skill_result(key, *, version=None)` | The same retrieval, reporting **why**: a frozen `SkillOutcome` with `.skill`, `.reason` (`ok` / `absent` / `integrity_failure` / `store_unavailable` / `wrong_version`), and `.detail`. Use it to fail closed on tampering — see *Failing closed on tampering* above. Raises only when no store is configured. | | `get_skills(refs)` | Batch form. Accepts `SkillReference` values and bare key strings (string = latest). Results follow input order; missing or unverifiable entries are omitted. | | `all_skills()` | Every verified skill the store holds, one per key at its newest version. | | `write_skills(skills, root, *, prune=True, timeout=10.0, on_unavailable="keep")` | Materialize skills under `root`, returning a `ReconcileReport`. `prune` removes formerly-managed skills no longer requested. `on_unavailable="raise"` raises instead of reporting when content cannot be retrieved. Raises `ValueError` for an unusable root, a negative `timeout`, or an unrecognised `on_unavailable`. **Performs synchronous filesystem I/O — see the note below.** | -| `SkillStore` | The structural interface content arrives through: `get_object(kind, key, version=None)`, `all_objects(kind)`, optional `is_initialized()`, `add_listener(kind, fn)` / `remove_listener(kind, fn)`. A store without `is_initialized()` is treated as initialized. | +| `SkillStore` | The structural interface content arrives through: `get_object(kind, key, version=None)`, `all_objects(kind)`, optional `is_initialized()`, `add_listener(kind, fn)` / `remove_listener(kind, fn)`. A store without `is_initialized()` is treated as initialized. Both shipped stores deliver only the skill kind, so `add_listener` on any other kind raises rather than being recorded and silently never firing. | | `InMemorySkillStore(objects=None)` | A dict-backed store with `put(raw)`, for local development and testing. Holds several versions of a key. | -| `FDv2SkillStore(sdk_key, *, base_uri=…, mode="stream", …)` | The delivery transport: a store fed by LaunchDarkly over the SDK-facing FDv2 channel. `start()`, `wait_for_skills(timeout)`, `is_initialized()`, `close()`, `diagnostics`, `failed`; also a context manager. **Server-side only** — a mobile key or client-side environment ID raises. See *Receiving skills from LaunchDarkly* above. | -| `watch_skills(skills, root, …)` | `write_skills` plus a re-reconcile on every delivery change. Returns `(initial report, SkillWatcher)`; close the watcher when done. Revocation then takes effect within `debounce` of arriving rather than at the next restart. | +| `FDv2SkillStore(sdk_key, *, base_uri=…, stream_uri=…, mode="stream", …)` | The delivery transport: a store fed by LaunchDarkly over the SDK-facing FDv2 channel. `start()`, `wait_for_skills(timeout)`, `is_initialized()`, `close()`, `diagnostics`, `failed`; also a context manager. `base_uri` and `stream_uri` are separate hosts, defaulting to LaunchDarkly's polling and streaming origins; `base_uri` alone covers both. `close()` is **final** — `start()` afterwards raises. **Server-side only** — a mobile key or client-side environment ID raises. See *Receiving skills from LaunchDarkly* above. | +| `watch_skills(skills, root, *, debounce=0.5, on_reconcile=None, …)` | `write_skills` plus a re-reconcile on every delivery change. Returns `(initial report, SkillWatcher)`; close the watcher when done. Revocation then takes effect within `debounce` of arriving rather than at the next restart. `debounce` is in **seconds** and must be non-negative and finite; `on_reconcile` is called with each *subsequent* report, the initial one being returned directly. One watcher per root. | | `StoreDiagnostics` | What the transport has seen: `payloads_transferred`, `skill_objects_received`, `objects_ignored`, `objects_revoked`, `hashless_objects`, `connection_failures`, `last_error`. | Configure the store with `init_client(options={"skillStore": store})`. With none configured, diff --git a/packages/client/agents.md b/packages/client/agents.md index 5533afb..d995575 100644 --- a/packages/client/agents.md +++ b/packages/client/agents.md @@ -673,11 +673,25 @@ it as soon as the write returns. **The platform bound is POSIX-only, and that is a decision — do not quietly "fix" it.** Windows reparse-point checks (`GetFileAttributesW`, `FILE_FLAG_OPEN_REPARSE_POINT`) are not implemented because Windows is not a supported or tested platform for this release: there is -no Windows CI runner in either repository, so the checks would ship unverified, and the -TypeScript SDK could not match them at all — Node exposes no `*at()` family on *any* -platform, so its racy floor is universal rather than Windows-only. Implementing them in -Python alone would break cross-language parity and trade a documented bound for an unverified -one. Two follow-on facts: on Windows write permission on the managed root is the only +no Windows CI runner in either repository, so the checks would ship unverified, and there is +no second implementation to check them against — the TypeScript SDK has no Windows story +either. Implementing them in Python alone would trade a documented bound for an unverified +one. + +The parity argument used to be stronger than that, and the correction matters because the +old wording is now wrong. It read: Node exposes no `*at()` family on *any* platform, so its +racy floor is universal rather than Windows-only. The first half is still true and the +second is not. `*at()` is not the only way to address a child relative to a pinned inode: +TypeScript commit `0a15b10` added a `SUPPORTS_PROC_FD` probe and `/proc/self/fd//` +addressing, which the Linux kernel resolves from the inode the descriptor holds rather than +from the name it was opened under. That **closes** the swap window on Linux exactly as +`*at()` does here, so TypeScript's `lstat` floor now applies on macOS and Windows only — +the same shape as this side's, not a universal one. + +None of which reopens the decision above. It never rested on TypeScript being equally +exposed; it rests on there being no Windows CI runner to verify the checks against, which is +still the case in both repositories. Two follow-on facts: on Windows write permission on the +managed root is the only boundary, which is why the privilege-separated deployment is documented as the mitigation rather than as advice; and this bound retroactively lowers the priority of the reserved-device-name work above — keep that code, but do not read it as evidence that Windows is hardened. If diff --git a/packages/client/src/launchdarkly_ai_server/skills.py b/packages/client/src/launchdarkly_ai_server/skills.py index c29bafc..d83dc2b 100644 --- a/packages/client/src/launchdarkly_ai_server/skills.py +++ b/packages/client/src/launchdarkly_ai_server/skills.py @@ -153,10 +153,21 @@ def add_listener(self, kind: str, fn: Callable[[dict[str, Any]], Any]) -> None: """ Registers *fn* to be called with each raw object ``put`` under *kind*. - Only ``kind == SKILL_OBJECT_KIND`` is ever notified, because ``put`` only - accepts skill objects; a listener registered under any other kind is - recorded and never fires. + Only ``kind == SKILL_OBJECT_KIND`` is ever notified, because ``put`` + only accepts skill objects — so a registration for any other kind + **raises** rather than being recorded and silently never firing. This is + the reason ``watch_skills`` refuses a store with no ``add_listener`` at + all: a listener that never fires looks exactly like one whose objects + never changed, and a store that accepted the registration has promised + something it cannot keep. ``FDv2SkillStore.add_listener`` refuses the + same way. """ + if kind != SKILL_OBJECT_KIND: + raise ValueError( + f"InMemorySkillStore notifies only {SKILL_OBJECT_KIND!r} " + f"changes, so a listener on {kind!r} would never fire. Register " + f"it on {SKILL_OBJECT_KIND!r}." + ) self._listeners.setdefault(kind, []).append(fn) def remove_listener(self, kind: str, fn: Callable[[dict[str, Any]], Any]) -> None: diff --git a/packages/client/src/launchdarkly_ai_server/skills_core.py b/packages/client/src/launchdarkly_ai_server/skills_core.py index 8da0ba8..3cc1751 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_core.py +++ b/packages/client/src/launchdarkly_ai_server/skills_core.py @@ -421,7 +421,15 @@ def verified_bytes( key: str, content: str | bytes, expected_hash: str, version: int ) -> VerifiedContent | VerificationFailure: """ - The whole content half of integrity verification: encode, size, hash. + The whole content half of integrity verification: size, encoding, hash. + + The order is fixed, so content failing two classes reports one determined + code: **size** before **encoding** before **hash**. Size first is what makes + over-cap content carrying a lone surrogate report ``over_size_cap`` rather + than ``not_utf8`` — without a fixed order that input's code is whichever + check the implementation happens to reach first, and §3.21's "one code per + failure class" rule cannot be tested against it. The shape checks that + precede all three are in ``verify_raw_skill``. Accepts either shape content arrives in. A wire-shaped ``str`` is UTF-8 encoded here, once — the only place that encode happens. ``bytes`` is an @@ -439,6 +447,7 @@ def verified_bytes( first one's verdict forward: that puts a "trust the value computed upstream" branch inside the one function whose job is not to. """ + encodable = True if isinstance(content, bytes): encoded = content else: @@ -447,17 +456,18 @@ def verified_bytes( except UnicodeEncodeError: # json.loads turns a "\ud800" escape into an unpaired surrogate, # which has no UTF-8 encoding — so there are no bytes the server - # could have hashed. Never reach for errors="surrogatepass": it - # would fabricate bytes that could satisfy the hash comparison. - reason = "content is not encodable as UTF-8" - record_integrity_failure( - key, - reason, - reason_code="not_utf8", - version=version, - expected_hash=expected_hash, - ) - return VerificationFailure(reason) + # could have hashed. + # + # These replacement bytes exist only to measure the content against + # the size cap, so that the *reported* failure follows the fixed + # order above. They are never hashed and never returned: the + # ``not_utf8`` branch below returns before the hash comparison, and + # nothing else reads ``encoded`` on this path. That is the whole + # reason ``errors="replace"`` is safe here and + # ``errors="surrogatepass"`` would not be anywhere — fabricated + # bytes that reached the comparison could satisfy it. + encodable = False + encoded = content.encode("utf-8", errors="replace") if len(encoded) > MAX_SKILL_CONTENT_BYTES: reason = ( @@ -473,6 +483,17 @@ def verified_bytes( ) return VerificationFailure(reason) + if not encodable: + reason = "content is not encodable as UTF-8" + record_integrity_failure( + key, + reason, + reason_code="not_utf8", + version=version, + expected_hash=expected_hash, + ) + return VerificationFailure(reason) + # sha256, lowercase hex, over the verbatim bytes — no canonicalization and # no content parsing of any kind anywhere in the integrity path. observed_hash = hashlib.sha256(encoded).hexdigest() diff --git a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py index c0435d1..d04cb89 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_fdv2.py +++ b/packages/client/src/launchdarkly_ai_server/skills_fdv2.py @@ -86,8 +86,20 @@ """ DEFAULT_BASE_URI = "https://sdk.launchdarkly.com" -"""Where the SDK-facing FDv2 endpoints live. Overridable for Federal and private -instances.""" +"""Where ``GET /sdk/poll`` is served. Overridable for Federal instances, private +instances, and relay deployments.""" + +DEFAULT_STREAM_URI = "https://stream.launchdarkly.com" +""" +Where ``GET /sdk/stream`` is served. + +LaunchDarkly serves streaming from a different host than polling, which is +why this is a second default rather than a path under ``DEFAULT_BASE_URI``. + +A *base_uri* given on its own applies to both endpoints, because a relay or a +private instance serving both from one host should need only one option; see +``FDv2SkillStore.__init__``. +""" POLL_PATH = "/sdk/poll" STREAM_PATH = "/sdk/stream" @@ -202,9 +214,9 @@ def _require_server_side_credential(sdk_key: str) -> None: """The only hosts a plain ``http://`` base URI may name: a local test double.""" -def _require_https_base_uri(base_uri: str) -> None: +def _require_https_uri(base_uri: str, option: str = "base_uri") -> None: """ - Refuses a base URI that would send the SDK key in cleartext. + Refuses a URI that would send the SDK key in cleartext. Every request carries the environment's server-side SDK key in ``Authorization``, so the transport is ``https://`` only. The one exception @@ -215,7 +227,7 @@ def _require_https_base_uri(base_uri: str) -> None: """ if not isinstance(base_uri, str) or not base_uri.strip(): raise ValueError( - "FDv2SkillStore requires an https:// base URI; none was given." + f"FDv2SkillStore requires an https:// URI for {option}; none was given." ) parts = urllib.parse.urlsplit(base_uri.strip()) if parts.scheme == "https" and parts.hostname: @@ -224,14 +236,15 @@ def _require_https_base_uri(base_uri: str) -> None: return if parts.scheme == "http": raise ValueError( - f"FDv2SkillStore refuses base_uri {base_uri!r}: a plain http:// URI " + f"FDv2SkillStore refuses {option} {base_uri!r}: a plain http:// URI " "would send the server-side SDK key in cleartext. Use https:// " - "(the default is https://sdk.launchdarkly.com). Plain http:// is " + "(the defaults are https://sdk.launchdarkly.com for polling and " + "https://stream.launchdarkly.com for streaming). Plain http:// is " "allowed only for a loopback host (localhost, 127.0.0.1, ::1) " "serving a local test double." ) raise ValueError( - f"FDv2SkillStore refuses base_uri {base_uri!r}: expected an https:// URI " + f"FDv2SkillStore refuses {option} {base_uri!r}: expected an https:// URI " "with a host, such as https://sdk.launchdarkly.com." ) @@ -686,8 +699,14 @@ def _delete_object(self, data: Any) -> _TransferOutcome: tombstone = _tombstone_from_delete(data) if tombstone is None: return _TransferOutcome() - target.delete(tombstone) - self.diagnostics.objects_revoked += 1 + removed = target.delete(tombstone) + if removed: + # Only a tombstone that took something away is a revocation. A + # delete for a key the store never held revoked nothing, and + # counting it inflates the one counter an operator reads to work + # out whether a revocation actually landed. ``changes`` below + # carries the tombstone either way, so a listener still sees it. + self.diagnostics.objects_revoked += 1 # A revocation identifies the skill payload just as a put does. self._skills_in_payload += 1 # A tombstone carries identity and no content; see @@ -879,6 +898,25 @@ def __init__(self, message: str, retry_after: float | None = None) -> None: self.retry_after = retry_after +class _StaleRequestStateError(_RecoverableTransportError): + """ + An HTTP 400 for a request carrying client state — the ``basis`` selector, or + an ``If-None-Match`` etag. + + That state is the one part of the request that can go stale, so it is + dropped and a full transfer asked for **once** before the status is treated + as fatal. The bound is what keeps a 400 from being plain "recoverable": a + request carrying no state at all was itself refused, and no reconnect fixes + that. + """ + + +_REQUEST_ADVICE = ( + "The request this adapter sent was not understood. It carries only the SDK " + "key and, after the first payload, a 'basis' selector, so check the base " + "URI and that the endpoint speaks FDv2." +) + _FORBIDDEN_ADVICE = ( "The FDv2 protocol is opt-in per LaunchDarkly account and is served as HTTP " "403 while it is off. Skill delivery needs it enabled; contact LaunchDarkly " @@ -930,12 +968,24 @@ def _classify_status(status: int, headers: Any) -> Exception: "base URI, and any proxy in between, for the address being redirected " "to." ) - if status in (400, 405, 406, 414, 501): + if status == 404: + # The endpoint does not exist for this credential or instance — a + # mistyped base URI, typically. No reconnect produces one. + return _FatalTransportError( + "LaunchDarkly returned HTTP 404 for the FDv2 endpoint. Check the " + "base URI, and that this instance serves /sdk/poll and /sdk/stream." + ) + if status == 400: + # The one rejection this adapter can act on: the selector it sent may be + # one the server no longer accepts. Recoverable so the state can be + # dropped and a full transfer requested; fatal once that has been tried. + return _StaleRequestStateError( + f"LaunchDarkly returned HTTP 400. {_REQUEST_ADVICE}" + ) + if status in (405, 406, 414, 501): return _FatalTransportError( f"LaunchDarkly returned HTTP {status}, which retrying will not fix. " - "The request this adapter sent was not understood. It carries only " - "the SDK key and, after the first payload, a 'basis' selector, so " - "check the base URI and that the endpoint speaks FDv2." + f"{_REQUEST_ADVICE}" ) return _RecoverableTransportError( f"LaunchDarkly returned HTTP {status}", _retry_after_seconds(headers) @@ -1038,10 +1088,15 @@ def __init__( base_uri: str, *, read_timeout: float, + stream_uri: str | None = None, opener: Any = None, ) -> None: self._sdk_key = sdk_key self._base_uri = base_uri.rstrip("/") + # Streaming and polling are served from different hosts by LaunchDarkly; + # a caller that names only one host means both, which is a relay or a + # private instance. ``FDv2SkillStore`` resolves the two-default case. + self._stream_uri = (stream_uri or base_uri).rstrip("/") self._read_timeout = read_timeout # Injectable, so an alternative transport can be supplied. The default # never follows a redirect; see ``_RefuseRedirects``. @@ -1066,25 +1121,28 @@ def interrupt(self) -> None: if response is not None: _interrupt_read(response) - def _url(self, path: str, basis: str | None) -> str: + def _url(self, origin: str, path: str, basis: str | None) -> str: """ The request URL: the path, plus ``basis`` once a payload has committed. + *origin* is the host for this path — polling and streaming have one + each. + Deliberately no ``mv`` (data model version). That parameter selects the *flag* data model and the connection rejects any value but the flag default; the agent-skill payload is generic, is served regardless of it, and has no model version of its own to ask for. """ if not basis: - return f"{self._base_uri}{path}" - return f"{self._base_uri}{path}?{urllib.parse.urlencode({'basis': basis})}" + return f"{origin}{path}" + return f"{origin}{path}?{urllib.parse.urlencode({'basis': basis})}" def _request( - self, path: str, basis: str | None, headers: dict[str, str] + self, origin: str, path: str, basis: str | None, headers: dict[str, str] ) -> urllib.request.Request: all_headers = {"Authorization": self._sdk_key, **headers} return urllib.request.Request( - self._url(path, basis), headers=all_headers, method="GET" + self._url(origin, path, basis), headers=all_headers, method="GET" ) def poll(self, basis: str | None, etag: str | None) -> _PollResult: @@ -1092,7 +1150,7 @@ def poll(self, basis: str | None, etag: str | None) -> _PollResult: headers = {"Accept": "application/json"} if etag: headers["If-None-Match"] = etag - request = self._request(POLL_PATH, basis, headers) + request = self._request(self._base_uri, POLL_PATH, basis, headers) try: with self._opener.open(request, timeout=self._read_timeout) as response: with self._lock: @@ -1126,6 +1184,7 @@ def poll(self, basis: str | None, etag: str | None) -> _PollResult: def stream(self, basis: str | None) -> _StreamConnection: """Opens ``GET /sdk/stream``.""" request = self._request( + self._stream_uri, STREAM_PATH, basis, {"Accept": "text/event-stream", "Cache-Control": "no-cache"}, @@ -1328,11 +1387,21 @@ class FDv2SkillStore: **Server-side only.** A mobile key or a client-side environment ID is refused in the constructor. - **The SDK key goes only where it was pointed.** *base_uri* must be - ``https://`` — plain ``http://`` is refused except to a loopback host, for - local test doubles — and redirects are never followed, so a 3xx from a proxy - or a private instance is a fatal failure rather than a request carrying the - key to whatever host ``Location`` named. + **The SDK key goes only where it was pointed.** *base_uri* and *stream_uri* + must each be ``https://`` — plain ``http://`` is refused except to a + loopback host, for local test doubles — and redirects are never followed, so + a 3xx from a proxy or a private instance is a fatal failure rather than a + request carrying the key to whatever host ``Location`` named. + + **Polling and streaming have separate hosts.** LaunchDarkly serves them from + different origins, so the defaults are a pair (``DEFAULT_BASE_URI`` and + ``DEFAULT_STREAM_URI``). A *base_uri* given on its own applies to both, + which is what a relay or a private instance serving both endpoints from one + host needs. + + **``close`` is final.** A closed store still answers from what it received, + but delivery cannot be resumed: ``start`` afterwards raises. Construct a new + store instead. **Delivery is in the background; retrieval is not.** A daemon thread owns the connection and fills memory, and ``get_object`` only ever reads what has @@ -1362,7 +1431,8 @@ def __init__( self, sdk_key: str, *, - base_uri: str = DEFAULT_BASE_URI, + base_uri: str | None = None, + stream_uri: str | None = None, mode: Mode = "stream", poll_interval: float = 30.0, read_timeout: float | None = None, @@ -1372,8 +1442,14 @@ def __init__( _requester: Any = None, ) -> None: """ - *base_uri* must be ``https://``; ``http://`` is accepted only for - ``localhost``, ``127.0.0.1`` or ``::1``. Raises ``ValueError`` otherwise. + *base_uri* is where ``GET /sdk/poll`` is sent (``DEFAULT_BASE_URI``) and + *stream_uri* where ``GET /sdk/stream`` is sent (``DEFAULT_STREAM_URI``), + because LaunchDarkly serves the two from different hosts. A *base_uri* + given **without** a *stream_uri* is used for both, which is what a relay + or a private instance serving both endpoints from one host needs; naming + both overrides them independently. Each must be ``https://``; + ``http://`` is accepted only for ``localhost``, ``127.0.0.1`` or + ``::1``. Raises ``ValueError`` otherwise. *mode* is ``"stream"`` by default. Prefer it: a ``delete-object`` reaches a live stream in seconds. ``"poll"`` exists for environments that cannot @@ -1392,10 +1468,19 @@ def __init__( *max_consecutive_failures* bounds the retry loop. On exceeding it the transport stops, logs an error, and the store keeps serving last known good; ``failed`` reports it. Only failures in a row count: a committed - payload resets the count. + payload resets the count. The one request built from nothing after a + stale selector is refused is exempt, so an outage that has already + spent the budget cannot swallow the one repair available. """ _require_server_side_credential(sdk_key) - _require_https_base_uri(base_uri) + # A lone ``base_uri`` means "both endpoints are here"; the two-host + # default applies only when neither was named. + if stream_uri is None: + stream_uri = DEFAULT_STREAM_URI if base_uri is None else base_uri + if base_uri is None: + base_uri = DEFAULT_BASE_URI + _require_https_uri(base_uri) + _require_https_uri(stream_uri, "stream_uri") if mode not in ("stream", "poll"): raise ValueError(f'mode must be "stream" or "poll", got {mode!r}') if poll_interval <= 0: @@ -1427,8 +1512,19 @@ def __init__( sdk_key.strip(), base_uri, read_timeout=read_timeout, + stream_uri=stream_uri, ) + self._closed = False + """ + ``close`` has been called. Final: ``start`` raises afterwards. + + What gives ``close`` a postcondition a caller can rely on — "delivery + has stopped" — including when the join timed out. A store that could be + restarted after a timed-out close leaves the caller unable to tell + whether delivery stopped, and a restart that silently never delivers + again is the failure this forecloses. To resume, construct a new store. + """ self._stop = threading.Event() self._first_payload = threading.Event() """A payload has committed. The fact ``wait_for_skills`` reports.""" @@ -1464,8 +1560,21 @@ def start(self) -> FDv2SkillStore: Starts the delivery thread. Idempotent; returns ``self`` so it chains. Does not block: use ``wait_for_skills`` when boot ordering matters. + + Raises ``RuntimeError`` on a **closed** store: ``close`` is final, so + there is no resuming it. A store that gave up at its failure bound is + not closed and can be started again — the retry budget and the terminal + reason both belong to the run that spent them. """ with self._lock: + if self._closed: + raise RuntimeError( + "This FDv2SkillStore has been closed, and close() is final: " + "delivery cannot be resumed, so a restarted store would " + "report itself started and never deliver. Construct a new " + "FDv2SkillStore to resume delivery. Held content is still " + "readable from the closed store." + ) # Read before the rearm clears it: a thread inside ``_give_up`` is # still alive and no longer delivering, so ``is_alive`` on its own # would have this call adopt a run that is about to return and @@ -1477,13 +1586,9 @@ def start(self) -> FDv2SkillStore: ) self._rearm_waiters() if delivering: - # A ``close`` whose join timed out leaves the previous thread - # running with the stop flag still set. Clearing it lets that - # thread carry on delivering, rather than leaving a store that - # reports itself started and never delivers again. - self._stop.clear() return self - self._stop.clear() + # Only ``close`` sets the stop flag, and a closed store never gets + # here, so there is nothing to clear. self._thread = threading.Thread( target=self._run, name="ld-ai-skills-fdv2", daemon=True ) @@ -1492,9 +1597,9 @@ def start(self) -> FDv2SkillStore: def _rearm_waiters(self) -> None: """ - Re-arms ``wait_for_skills`` for a store being started again after a - ``close``. A payload already held stays an answer; an ended delivery - does not, or the next waiter would be released before it began. + Re-arms ``wait_for_skills`` for a store being started again after it + gave up. A payload already held stays an answer; an ended delivery does + not, or the next waiter would be released before it began. A terminal ``failed`` reason is dropped for the same reason: it says why delivery stopped for good, and delivery is about to run again. Leaving @@ -1512,12 +1617,21 @@ def _rearm_waiters(self) -> None: def close(self, timeout: float = 5.0) -> None: """ - Stops delivery. Idempotent, and safe to call from any thread. + Stops delivery. Idempotent, safe to call from any thread, and **final**: + a subsequent ``start`` raises rather than resuming. Construct a new + store to resume delivery. + + Finality is what gives this call a postcondition — delivery has + stopped — even when the join below times out on a thread parked + somewhere no interrupt reaches. A store that could be restarted from + there would leave the caller unable to tell whether delivery stopped. Held content is *not* dropped: a closed store still answers from what it received. Detaching the store from the accessors is the job of the package-level ``launchdarkly_ai_server.shutdown()`` coroutine. """ + with self._lock: + self._closed = True self._stop.set() # A waiter parked in ``wait_for_skills`` is owed an answer now rather # than at the end of its timeout; delivery is over either way. @@ -1626,7 +1740,21 @@ def add_listener(self, kind: str, fn: Callable[[dict[str, Any]], Any]) -> None: *fn* runs on the delivery thread. Keep it cheap and non-blocking. An exception it raises is logged and swallowed, because a broken listener must not be able to kill delivery. + + Only ``SKILL_OBJECT_KIND`` is ever delivered, so a registration for any + other kind **raises** rather than being recorded and silently never + firing. ``InMemorySkillStore.add_listener`` refuses the same way, for + the reason ``watch_skills`` refuses a store with no ``add_listener`` at + all: a listener that never fires looks exactly like one whose objects + never changed, and a store that accepted the registration has promised + something it cannot keep. """ + if kind != SKILL_OBJECT_KIND: + raise ValueError( + f"FDv2SkillStore notifies only {SKILL_OBJECT_KIND!r} changes, so " + f"a listener on {kind!r} would never fire. Register it on " + f"{SKILL_OBJECT_KIND!r}." + ) with self._lock: self._listeners.setdefault(kind, []).append(fn) @@ -1685,6 +1813,22 @@ def _run(self) -> None: # would spend a retry from the bounded budget and leave a # misleading ``last_error`` on a healthy store. return + repairing_state = False + if isinstance(exc, _StaleRequestStateError): + # The selector and the etag are the only client state in the + # request, so a rejection of a request carrying neither is + # the request itself being refused, and reconnecting cannot + # fix it. Carrying one, the state may be stale: drop it, ask + # for a full transfer, and let the next 400 be the fatal one. + with self._lock: + exhausted = self._basis is None and self._etag is None + if not exhausted: + self._basis = None + self._etag = None + repairing_state = True + if exhausted: + self._give_up(str(exc)) + return with self._lock: # Whatever the dropped connection had transferred so far is # not a payload; the next connection starts one afresh. @@ -1694,7 +1838,13 @@ def _run(self) -> None: answered = self._attempt_answered self._reader.diagnostics.connection_failures = failures self._reader.diagnostics.last_error = str(exc) - if failures > self._max_consecutive_failures: + if failures > self._max_consecutive_failures and not repairing_state: + # The one-shot request built from nothing is exempt from the + # bound, so an outage that has already spent the budget + # cannot swallow the repair a stale selector is asking for. + # It cannot unbound the loop either: the repaired request + # carries no state, so a second 400 is fatal on its own and + # any other failure meets a budget still over the bound. self._give_up( f"gave up after {failures} consecutive failures; " f"last error: {exc}" @@ -1705,6 +1855,14 @@ def _run(self) -> None: delay = _backoff_delay( failures, base=self._initial_backoff, maximum=self._max_backoff ) + else: + # A server asking for no delay still gets one: honouring + # ``Retry-After: 0`` literally would reconnect as fast as + # the loop allows and burn the whole retry bound in + # milliseconds, hammering the endpoint on the way. The + # floor is ``initial_backoff``, the same floor our own + # backoff starts from. + delay = max(delay, self._initial_backoff) # ``Retry-After`` is a request and ``max_backoff`` is a promise. # The header may come from a proxy rather than LaunchDarkly, and # a value in the hours would park revocation for that long. diff --git a/packages/client/src/launchdarkly_ai_server/skills_watch.py b/packages/client/src/launchdarkly_ai_server/skills_watch.py index 72cad73..5ee8649 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_watch.py +++ b/packages/client/src/launchdarkly_ai_server/skills_watch.py @@ -24,6 +24,7 @@ import asyncio import logging +import math import os import threading from collections.abc import Callable, Sequence @@ -267,6 +268,10 @@ async def watch_skills( never changed. The optional ``remove_listener`` lets ``SkillWatcher.close`` detach from the store; a store without it still works, but each closed watcher then stays registered for the store's lifetime. + + *debounce* must be a non-negative finite number of seconds. ``NaN`` raises + alongside a negative value: it would pass a bare ``< 0`` guard and then + collapse the coalescing window to nothing. """ store = get_store() if store is None: @@ -282,8 +287,16 @@ async def watch_skills( "observed. Use write_skills for a one-shot reconcile, or configure a " "store with a delivery transport (FDv2SkillStore)." ) - if debounce < 0: - raise ValueError(f"debounce must not be negative, got {debounce!r}") + # ``NaN`` is the case a bare ``< 0`` guard misses: ``nan < 0`` is ``False``, + # so it passes validation and then collapses the window to nothing, because + # ``Event.wait(nan)`` returns immediately. Every delivered object would + # reconcile on its own with no coalescing at all — the opposite of what the + # option is for. ``write_skills`` already guards its ``timeout`` this way. + if not math.isfinite(debounce) or debounce < 0: + raise ValueError( + f"debounce must be a non-negative finite number of seconds, got " + f"{debounce!r}" + ) # The watcher attaches its listener before the initial reconcile, not after. # The reconcile snapshots the store as its first step and then spends the diff --git a/packages/client/src/launchdarkly_ai_server/types_validation.py b/packages/client/src/launchdarkly_ai_server/types_validation.py index 37cd56b..77ad601 100644 --- a/packages/client/src/launchdarkly_ai_server/types_validation.py +++ b/packages/client/src/launchdarkly_ai_server/types_validation.py @@ -156,9 +156,15 @@ def parse_ai_config(raw: Any) -> ParseResult: error={"message": "outputFormat must be an object (JSON Schema)"}, ) - skills = raw.get("skills") - if skills is not None: - err = _parse_skills(skills) + # ``in`` rather than ``is not None``: an explicit ``skills: null`` must fail + # the parse, not read as absent. Read as absent it makes ``skill_refs`` + # return ``[]``, and a ``prune=True`` reconcile then *deletes* previously + # materialized skill files on the strength of a field the SDK could not + # parse — the hazard the store path already refuses, where reading a + # malformed object as absent would let prune delete the last known-good + # copy on disk. Failing the whole parse is the louder and safer outcome. + if "skills" in raw: + err = _parse_skills(raw["skills"]) if err: return ParseFailure(success=False, error={"message": err}) diff --git a/packages/client/tests/test_schema.py b/packages/client/tests/test_schema.py index 3be146d..359a890 100644 --- a/packages/client/tests/test_schema.py +++ b/packages/client/tests/test_schema.py @@ -124,6 +124,33 @@ def test_key_at_length_bound_accepted(self) -> None: def test_non_array_skills_fails(self, bad_skills: Any) -> None: assert parse_ai_config(self._base(skills=bad_skills)).success is False + def test_an_explicit_null_skills_fails_rather_than_reading_as_absent( + self, + ) -> None: + """``skills: null`` is the non-array that reads as "no skills". + + It must not be treated that way. Read as absent it makes ``skill_refs`` + return ``[]``, and a ``prune=True`` reconcile then *deletes* previously + materialized skill files on the strength of a field the SDK could not + parse — the same hazard the store path refuses, where reading a + malformed object as absent would let prune delete the last known-good + copy on disk. Failing the whole parse is the louder and safer outcome, + and it is why this case is pinned apart from the other non-arrays. + """ + result = parse_ai_config(self._base(skills=None)) + assert result.success is False + assert "skills" in result.error["message"] + + def test_an_absent_skills_key_is_still_valid(self) -> None: + """The other half of the bullet above: *absent* is not *null*. + + Rejecting ``null`` must not cost backward compatibility for the + configs that simply have no ``skills`` field. + """ + raw = self._base() + assert "skills" not in raw + assert parse_ai_config(raw).success is True + @pytest.mark.parametrize("entry", ["pdf-extraction", 1, None, ["a", 1]]) def test_non_object_entry_fails(self, entry: Any) -> None: assert parse_ai_config(self._base(skills=[entry])).success is False diff --git a/packages/client/tests/test_skills.py b/packages/client/tests/test_skills.py index 1b8ebd5..15a885a 100644 --- a/packages/client/tests/test_skills.py +++ b/packages/client/tests/test_skills.py @@ -555,16 +555,20 @@ def test_put_notifies_skill_kind_listeners(self, make_raw_skill: Any) -> None: assert seen == [raw] - def test_put_does_not_notify_other_kind_listeners( - self, make_raw_skill: Any - ) -> None: + def test_add_listener_for_a_non_skill_kind_raises(self) -> None: + """A listener that can never fire is refused, not recorded. + + ``put`` only accepts skill objects, so nothing else is ever delivered: + a recorded listener on another kind would silently never fire, and that + is indistinguishable from one whose objects never changed. It is the + same failure §3.26 refuses when the store has no ``add_listener`` at + all, so it gets the same loud answer. + """ s = InMemorySkillStore() - seen: list[dict[str, Any]] = [] - s.add_listener("flag", seen.append) - - s.put(make_raw_skill(key="a")) - - assert seen == [] + with pytest.raises(ValueError, match="would never fire") as excinfo: + s.add_listener("flag", print) + assert "skill" in str(excinfo.value) + assert s._listeners == {} def test_remove_listener_stops_put_notifying_it(self, make_raw_skill: Any) -> None: s = InMemorySkillStore() @@ -1727,6 +1731,31 @@ async def test_one_record_per_reason_code( # a well-formed key and digest never enter a redaction branch. assert "Do the thing." not in json.dumps(record) + async def test_over_cap_content_with_a_lone_surrogate_reports_the_size( + self, caplog: pytest.LogCaptureFixture + ) -> None: + """The one input on which the check order is observable. + + Content that is both over the cap and unencodable fails two classes, so + without a fixed order its code is whichever check the implementation + reaches first — and the "one code per failure class" rule above cannot + be tested against it. The order is shape, then **size**, then + **encoding**, then hash: size is the cheap check, and running an + encoding pass over a 10 MiB body before rejecting it for being 10 MiB + is a DoS foothold rather than a nicety. + """ + from launchdarkly_ai_server import skills_core + + body = _OVERSIZE + json.loads(r'"\ud800"') + await self._withhold({"a": _raw_object(content=body)}) + + records = _integrity_records(caplog) + assert len(records) == 1 + assert records[0]["reason_code"] == "over_size_cap" + # The cap is named, interpolated from the constant rather than restated, + # so raising it cannot leave a stale figure in the message. + assert str(skills_core.MAX_SKILL_CONTENT_BYTES) in records[0]["reason"] + def test_the_case_table_exhausts_the_vocabulary(self) -> None: """The vocabulary is closed, and every token in it is reachable. diff --git a/packages/client/tests/test_skills_fdv2.py b/packages/client/tests/test_skills_fdv2.py index d7affd6..5ec48bf 100644 --- a/packages/client/tests/test_skills_fdv2.py +++ b/packages/client/tests/test_skills_fdv2.py @@ -42,12 +42,15 @@ ) from launchdarkly_ai_server.skills_core import SKILL_OBJECT_KIND from launchdarkly_ai_server.skills_fdv2 import ( + DEFAULT_BASE_URI, DEFAULT_POLL_TIMEOUT, DEFAULT_STREAM_READ_TIMEOUT, + DEFAULT_STREAM_URI, FDV2_KEY_DELIMITER, FDV2_OBJECT_KIND, MAX_RESPONSE_BYTES, _backoff_delay, + _classify_status, _FatalTransportError, _is_skill_event, _iter_sse, @@ -56,6 +59,7 @@ _Requester, _retry_after_seconds, _SkillObjectSet, + _StaleRequestStateError, _store_object_from_put, _StreamConnection, _tombstone_from_delete, @@ -637,6 +641,33 @@ def test_a_delete_object_revokes_the_skill(self) -> None: assert held.get("pdf-extraction", None) is None assert reader.diagnostics.objects_revoked == 1 + def test_a_delete_for_a_key_never_held_is_not_counted_as_a_revocation( + self, + ) -> None: + """``objects_revoked`` counts what went away, not tombstones seen. + + The counter is operator-facing, and it is read precisely when somebody + is working out whether a revocation landed. A delete for a key the store + never held revoked nothing, so counting it inflates the one number that + answers that question. The tombstone still reaches listeners through + ``changes``, which is where "every revocation the server stated" lives. + """ + held = _SkillObjectSet() + reader = _ProtocolReader(held) + drive(reader, full_payload(("put-object", put_skill(key="kept")))) + outcomes = drive( + reader, + events( + ("server-intent", server_intent("xfer-changes")), + ("delete-object", delete_skill(key="never-delivered")), + ("payload-transferred", transferred("basis-2")), + ), + ) + assert reader.diagnostics.objects_revoked == 0 + assert held.get("kept", None) is not None + # Reported, just not counted. + assert outcomes[-1].changes == [{"key": "never-delivered", "version": 3}] + def test_a_delete_notifies_with_a_tombstone_carrying_no_content(self) -> None: held = _SkillObjectSet() reader = _ProtocolReader(held) @@ -1613,6 +1644,126 @@ def test_a_restart_returns_the_retry_budget(self, endpoint: Any) -> None: assert store.wait_for_skills(timeout=5) is True assert store.failed is None + def test_a_404_stops_delivery_immediately(self, endpoint: Any) -> None: + """A 404 means the endpoint does not exist for this credential. + + A mistyped base URI, typically, or an instance that does not serve the + FDv2 endpoints. No reconnect produces one, so it is fatal rather than + retried — and it is the exception to the shape the recoverable-by-default + rule would suggest. + """ + endpoint.queue_poll(status=404) + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with poll_store(endpoint) as store: + assert wait_until(lambda: store.failed is not None) + assert store.wait_for_skills(timeout=1) is False + assert "404" in store.failed + assert "/sdk/poll" in store.failed + # Fatal means one request, not a retry that happened to find the payload. + assert len(endpoint.requests) == 1 + assert store.diagnostics.connection_failures == 0 + + def test_a_400_reconnects_once_from_scratch_and_is_then_fatal( + self, endpoint: Any + ) -> None: + """A 400 is what a stale ``basis`` selector looks like. + + The selector and the etag are the only client state the request carries, + so a fresh connection built from nothing is the one repair available. + It gets exactly one: the retry bound is what keeps this from being + "400 is recoverable". + """ + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + endpoint.queue_poll(status=400) + endpoint.queue_poll(status=400) + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with poll_store(endpoint) as store: + assert store.wait_for_skills(timeout=5) is True + assert wait_until(lambda: store.failed is not None) + assert "400" in store.failed + # The first payload, then the retried request, then the fatal one. The + # fourth queued payload is never asked for. + assert len(endpoint.requests) == 3 + # The premise: the rejected request did carry client state to drop. + assert "basis" in endpoint.requests[1]["query"] + # The retry was from scratch: no selector and no etag on the way back. + retried = endpoint.requests[2] + assert retried["query"] == {} + assert retried["if_none_match"] is None + # Last known good survives both. + assert store.get_object(SKILL_OBJECT_KIND, "pdf-extraction") is not None + + def test_a_400_still_reconnects_from_scratch_on_a_spent_budget( + self, endpoint: Any + ) -> None: + """The one repair available does not compete with the retry bound. + + A 400 arriving on a budget an outage has already spent would otherwise + give up while holding the one request known to fix it, and delivery + would stop for the process lifetime over state the store was about to + drop. Exempting that request cannot unbound the loop: it carries no + state, so a second 400 is fatal on its own. + """ + store = poll_store(endpoint, max_consecutive_failures=1) + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + endpoint.queue_poll(status=500) + endpoint.queue_poll(status=400) + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with store: + assert store.wait_for_skills(timeout=5) is True + # The premise: the budget is spent by the time the 400 arrives. + assert wait_until(lambda: len(endpoint.requests) == 4) + assert store.failed is None + # The repair went out from scratch rather than never going out at all. + repair = endpoint.requests[3] + assert repair["query"] == {} + assert repair["if_none_match"] is None + + def test_a_non_400_after_the_repair_meets_the_spent_budget( + self, endpoint: Any + ) -> None: + """The exemption is for the repair, not for the run that follows it.""" + store = poll_store(endpoint, max_consecutive_failures=1) + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + endpoint.queue_poll(status=500) + endpoint.queue_poll(status=400) + endpoint.queue_poll(status=500) + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with store: + assert store.wait_for_skills(timeout=5) is True + assert wait_until(lambda: store.failed is not None) + assert "gave up after 3 consecutive failures" in store.failed + # The fifth queued payload is never asked for. + assert len(endpoint.requests) == 4 + + def test_a_400_carrying_no_client_state_is_fatal_at_once( + self, endpoint: Any + ) -> None: + """There is nothing to drop on a first connection, so nothing to repair. + + A request that carried neither a selector nor an etag and was still + refused was refused on its own terms. + """ + endpoint.queue_poll(status=400) + endpoint.queue_poll(full_payload(("put-object", put_skill()))) + with poll_store(endpoint) as store: + assert wait_until(lambda: store.failed is not None) + assert "400" in store.failed + assert len(endpoint.requests) == 1 + + def test_the_two_exceptional_statuses_are_classified_apart(self) -> None: + # The classification is the contract; the end-to-end tests above are + # what prove the loop honours it. + assert isinstance(_classify_status(404, None), _FatalTransportError) + assert isinstance(_classify_status(400, None), _StaleRequestStateError) + # A stale-state error is still a recoverable one, so the retry path + # reaches it at all. + assert isinstance(_classify_status(400, None), _RecoverableTransportError) + for status in (405, 406, 414, 501): + assert isinstance(_classify_status(status, None), _FatalTransportError) + assert isinstance(_classify_status(503, None), _RecoverableTransportError) + assert not isinstance(_classify_status(503, None), _StaleRequestStateError) + def test_a_401_stops_delivery(self, endpoint: Any) -> None: endpoint.queue_poll(status=401) with poll_store(endpoint) as store: @@ -1812,31 +1963,68 @@ def test_stream_retries_are_bounded(self) -> None: def test_a_retry_after_header_is_honoured(self) -> None: requester = _ScriptedRequester( - _RecoverableTransportError("slow down", retry_after=0.25), + _RecoverableTransportError("slow down", retry_after=0.5), ) store = FDv2SkillStore( SDK_KEY, mode="poll", poll_interval=10.0, - initial_backoff=5.0, + initial_backoff=0.01, + max_backoff=5.0, _requester=requester, ) try: started = time.monotonic() store.start() - assert wait_until(lambda: len(requester.calls) >= 2, timeout=3) + assert wait_until(lambda: len(requester.calls) >= 2, timeout=5) elapsed = time.monotonic() - started - # The server asked for 0.25s; our own backoff would have been 5s. - assert 0.2 <= elapsed < 3.0 + # The server asked for 0.5s and our own backoff would have been + # 0.01s, so waiting is the only way the header could have been read. + # Asked *longer* rather than shorter on purpose: a shorter request + # is floored at ``initial_backoff``, so it cannot discriminate. + assert elapsed >= 0.4 + finally: + store.close() + + def test_a_retry_after_of_zero_still_waits_the_initial_backoff(self) -> None: + """``Retry-After: 0`` is floored, not taken literally. + + A server — or an intermediate proxy — answering ``0`` would otherwise + have the loop reconnect as fast as it can schedule, spending the whole + bounded retry budget in milliseconds and hammering the endpoint on the + way. The floor is ``initial_backoff``, the same floor our own backoff + starts from. + """ + requester = _ScriptedRequester( + _RecoverableTransportError("slow down", retry_after=0.0), + ) + store = FDv2SkillStore( + SDK_KEY, + mode="poll", + poll_interval=10.0, + initial_backoff=0.5, + max_backoff=5.0, + _requester=requester, + ) + try: + started = time.monotonic() + store.start() + assert wait_until(lambda: len(requester.calls) >= 2, timeout=5) + assert time.monotonic() - started >= 0.4 finally: store.close() def test_a_retry_after_header_is_parsed_off_the_wire(self, endpoint: Any) -> None: - endpoint.queue_poll(status=429, retry_after="0") + endpoint.queue_poll(status=429, retry_after="0.5") endpoint.queue_poll(full_payload(("put-object", put_skill()))) - with poll_store(endpoint, initial_backoff=5.0) as store: - # If Retry-After were ignored the 5s backoff would blow the timeout. - assert store.wait_for_skills(timeout=3) is True + started = time.monotonic() + with poll_store( + endpoint, initial_backoff=0.01, max_backoff=5.0, poll_interval=10.0 + ) as store: + assert store.wait_for_skills(timeout=5) is True + # 0.5s is only obtainable from the header: our own backoff here is 0.01s + # and the cap is 5s, so neither could have produced this wait. + assert time.monotonic() - started >= 0.4 @pytest.mark.parametrize("raw", ["inf", "Infinity", "-inf", "nan", "1e309"]) def test_a_non_finite_retry_after_is_ignored(self, raw: str) -> None: @@ -2543,6 +2731,80 @@ def test_https_is_accepted(self) -> None: assert FDv2SkillStore(SDK_KEY, base_uri="https://sdk.example.com/") is not None assert FDv2SkillStore(SDK_KEY) is not None + def test_a_plain_http_stream_uri_is_refused_by_name(self) -> None: + """The streaming host is checked too, and the message names it.""" + with pytest.raises(ValueError, match="cleartext") as excinfo: + FDv2SkillStore( + SDK_KEY, + base_uri="https://sdk.example.com", + stream_uri="http://stream.example.com", + ) + assert "stream_uri" in str(excinfo.value) + + +class TestStreamHostDefaults: + """ + LaunchDarkly serves ``/sdk/stream`` from a different host than ``/sdk/poll``, + and ``mode="stream"`` is the default — so a single-host default would have + the *default* configuration talk to the wrong host on first contact with a + real environment. Both base server-side SDKs ship the hosts as a pair + (``ldclient.Config``'s ``stream_uri``, ``js-server-sdk-common``'s + ``streamUri``), which is what these defaults follow. + """ + + @staticmethod + def _origins(store: FDv2SkillStore) -> tuple[str, str]: + requester = store._requester + return requester._base_uri, requester._stream_uri + + def test_the_two_defaults_are_different_hosts(self) -> None: + assert DEFAULT_BASE_URI == "https://sdk.launchdarkly.com" + assert DEFAULT_STREAM_URI == "https://stream.launchdarkly.com" + assert self._origins(FDv2SkillStore(SDK_KEY)) == ( + DEFAULT_BASE_URI, + DEFAULT_STREAM_URI, + ) + + def test_a_base_uri_alone_serves_both_endpoints(self) -> None: + # A relay or a private instance serving both from one host needs one + # option, not two. + assert self._origins( + FDv2SkillStore(SDK_KEY, base_uri="https://relay.example.com") + ) == ("https://relay.example.com", "https://relay.example.com") + + def test_naming_both_overrides_them_independently(self) -> None: + assert self._origins( + FDv2SkillStore( + SDK_KEY, + base_uri="https://sdk.example.com/", + stream_uri="https://stream.example.com/", + ) + ) == ("https://sdk.example.com", "https://stream.example.com") + + def test_a_stream_uri_alone_leaves_polling_on_its_default(self) -> None: + assert self._origins( + FDv2SkillStore(SDK_KEY, stream_uri="https://stream.example.com") + ) == (DEFAULT_BASE_URI, "https://stream.example.com") + + def test_the_stream_request_goes_to_the_stream_host( + self, endpoint: Any, second_endpoint: Any + ) -> None: + """The split is on the wire, not only in the attributes. + + Polling at one host and streaming at another is the whole point, so + assert it where it is observable: the streaming request arrives at the + streaming endpoint and nothing arrives at the polling one. + """ + requester = _Requester( + SDK_KEY, + endpoint.base_uri, + read_timeout=5.0, + stream_uri=second_endpoint.base_uri, + ) + requester.stream(None).close() + assert [r["path"] for r in second_endpoint.requests] == ["/sdk/stream"] + assert endpoint.requests == [] + # --------------------------------------------------------------------------- # The eager re-reconcile, end to end over the transport @@ -2649,6 +2911,19 @@ class TestListenerRegistration: def _skill_listeners(store: Any) -> list[Any]: return list(store._listeners.get(SKILL_OBJECT_KIND, [])) + def test_fdv2_add_listener_for_a_non_skill_kind_raises(self, endpoint: Any) -> None: + """The transport refuses the same registration the in-memory store does. + + Only skill objects are ever delivered here, so a listener on any other + kind would never fire — and a store that accepted it has promised + something it cannot keep. + """ + with poll_store(endpoint) as store: + with pytest.raises(ValueError, match="would never fire") as excinfo: + store.add_listener("flag", print) + assert SKILL_OBJECT_KIND in str(excinfo.value) + assert store._listeners == {} + def test_fdv2_remove_listener_of_an_unregistered_callable_is_a_no_op( self, endpoint: Any ) -> None: @@ -2683,6 +2958,42 @@ def test_close_is_idempotent(self, endpoint: Any) -> None: store.close() store.close() + def test_a_closed_store_does_not_restart(self, endpoint: Any) -> None: + """``close`` is final, and a restart raises rather than resuming. + + Finality is what gives ``close`` a postcondition a caller can rely on — + delivery has stopped — including when the join timed out. A store that + could be restarted from there leaves the caller unable to tell whether + delivery stopped, and a restart that silently never delivered again is + the failure this forecloses. To resume, construct a new store. + """ + store = poll_store(endpoint) + store.start() + store.close() + with pytest.raises(RuntimeError, match="close\\(\\) is final") as excinfo: + store.start() + # The remedy is in the message, not only in the docs. + assert "Construct a new FDv2SkillStore" in str(excinfo.value) + + def test_a_store_closed_before_it_started_also_refuses_to_start( + self, endpoint: Any + ) -> None: + store = poll_store(endpoint) + store.close() + with pytest.raises(RuntimeError, match="close\\(\\) is final"): + store.start() + + def test_reentering_a_closed_store_as_a_context_manager_raises( + self, endpoint: Any + ) -> None: + # ``__enter__`` is ``start``, so finality reaches the ``with`` form too. + store = poll_store(endpoint) + with store: + pass + with pytest.raises(RuntimeError, match="close\\(\\) is final"): + with store: + pass + def test_close_during_a_slow_connect_returns_promptly(self) -> None: # Before the connect returns there is no connection for close() to # interrupt. If the delivery thread then enters the read anyway, close() @@ -2975,19 +3286,36 @@ def test_a_payload_already_held_still_answers_true_after_close(self) -> None: store.close() assert store.wait_for_skills(timeout=5) is True - def test_a_restarted_store_waits_again(self) -> None: - # The released flag is sticky by design, so a store closed before any - # payload and then started again has to re-arm: otherwise the next - # waiter is let go before delivery has had a chance to begin. - store = stream_store(_requester=_SilentStreamRequester()) + def test_a_store_restarted_after_giving_up_waits_again(self) -> None: + # The released flag is sticky by design, so a store that gave up before + # any payload and is then started again has to re-arm: otherwise the + # next waiter is let go before delivery has had a chance to begin. + # Restarting after a *close* is not available — see + # ``TestCloseIsFinal`` — so the give-up path is what exercises this. + class _FailsThenGoesQuiet(_FakeRequester): + """One failure, enough to give up; silent on every run after.""" + + def __init__(self) -> None: + self.attempts = 0 + + def stream(self, basis: str | None) -> Any: + self.attempts += 1 + if self.attempts == 1: + raise _RecoverableTransportError("x") + return _BlockingConnection() + + store = stream_store( + max_consecutive_failures=0, _requester=_FailsThenGoesQuiet() + ) store.start() - store.close() + assert wait_until(lambda: store.failed is not None) assert store.wait_for_skills(timeout=0.1) is False + store.start() try: started = time.monotonic() assert store.wait_for_skills(timeout=0.5) is False - # Waited, rather than being released by the previous close. + # Waited, rather than being released by the previous give-up. assert time.monotonic() - started >= 0.4 finally: store.close() @@ -3066,11 +3394,12 @@ def test_a_poll_we_interrupted_is_not_a_delivery_failure( assert store.diagnostics.last_error is None assert store.failed is None - def test_a_close_that_timed_out_leaves_the_store_restartable(self) -> None: + def test_a_close_that_timed_out_is_still_final(self) -> None: # A request blocked inside its connect is beyond any interrupt, so - # ``close`` can still return with the thread alive. ``start`` must not - # then find that thread and return with the stop flag set: the store - # would report itself started and never deliver again. + # ``close`` can still return with the thread alive. This is the case + # finality exists for: the caller cannot tell whether delivery stopped, + # and a ``start`` that adopted the dying thread would leave a store + # reporting itself started and never delivering. Raising says so. requester = _SlowPollRequester() store = FDv2SkillStore( SDK_KEY, mode="poll", poll_interval=0.01, _requester=requester @@ -3080,8 +3409,8 @@ def test_a_close_that_timed_out_leaves_the_store_restartable(self) -> None: assert requester.entered.wait(timeout=5) store.close(timeout=0.2) assert store._thread is not None and store._thread.is_alive() - store.start() - assert store._stop.is_set() is False + with pytest.raises(RuntimeError, match="close\\(\\) is final"): + store.start() finally: requester.release.set() store.close(timeout=2) diff --git a/packages/client/tests/test_skills_watch.py b/packages/client/tests/test_skills_watch.py index 4c30f88..5c60d06 100644 --- a/packages/client/tests/test_skills_watch.py +++ b/packages/client/tests/test_skills_watch.py @@ -60,22 +60,185 @@ async def test_the_in_memory_store_can_also_drive_a_watch( finally: watcher.close() - async def test_a_burst_of_changes_coalesces_into_few_reconciles( + async def test_a_burst_of_changes_coalesces_into_a_debounce_window( self, tmp_path: Any, make_raw_skill: Any ) -> None: + """Six notifications spread across one window produce two reconciles. + + Three things are needed for this to discriminate rather than pass + vacuously, and this test does all three: + + * The puts happen **after** ``watch_skills`` has attached its listener. + A payload committed before that reaches nobody, the counter stays at + zero, and an upper bound then passes against an implementation with + the debouncing deleted. + * The counter is asserted to have **moved**, and against an exact + figure. A run in which nothing was ever notified is not a test of + coalescing. + * The notifications are **spread over time**. A burst arriving inside + one synchronous pass of the listener collapses to a single reconcile + whether or not the debounce exists, because the worker had not woken + yet — so a payload of twelve objects put back to back proves nothing. + Each put here lands in its own scheduler pass. + + **Two, not one.** This watcher clears its wake flag before the window + rather than after, so a change arriving *inside* the window schedules a + further pass instead of being merged into the reconcile that is about to + run: six puts across one window are one reconcile for the window plus + one for the spill. Six separate reconciles is what the debounce + prevents, and is what this figure discriminates against. + """ store = InMemorySkillStore() await init_client(options={"skillStore": store}, client=object()) - _report, watcher = await watch_skills("*", tmp_path / "s", debounce=0.1) + _report, watcher = await watch_skills("*", tmp_path / "s", debounce=0.6) try: - for i in range(12): + assert watcher.reconciles == 0 + for i in range(6): store.put(make_raw_skill(key=f"skill-{i}")) - time.sleep(0.5) - # Twelve objects put back to back fire twelve listener calls; without - # coalescing that is twelve reconciles of one root. - assert watcher.reconciles <= 2 + time.sleep(0.1) + assert wait_until(lambda: watcher.reconciles >= 2, timeout=10) + # Settle well past two windows, so a third would have landed by now. + time.sleep(1.5) + assert watcher.reconciles == 2 finally: watcher.close() + async def test_changes_spread_beyond_the_window_do_not_coalesce( + self, tmp_path: Any, make_raw_skill: Any + ) -> None: + """The control for the test above: the window is what merges them. + + Without it the two tests could both be satisfied by a watcher that + reconciles once and then never again. + """ + store = InMemorySkillStore() + await init_client(options={"skillStore": store}, client=object()) + _report, watcher = await watch_skills("*", tmp_path / "s", debounce=0.05) + try: + store.put(make_raw_skill(key="one")) + assert wait_until(lambda: watcher.reconciles >= 1, timeout=10) + store.put(make_raw_skill(key="two")) + assert wait_until(lambda: watcher.reconciles >= 2, timeout=10) + finally: + watcher.close() + + @pytest.mark.parametrize( + "debounce", [-1.0, -0.001, float("nan"), float("inf"), float("-inf")] + ) + async def test_a_negative_or_non_finite_debounce_raises( + self, tmp_path: Any, debounce: float + ) -> None: + """``NaN`` is the case a bare ``< 0`` guard misses. + + ``nan < 0`` is ``False``, so it would pass validation and then collapse + the window to nothing — ``Event.wait(nan)`` returns immediately, so + every delivered object reconciles on its own with no coalescing at all. + ``write_skills`` already guards its ``timeout`` this way. + """ + store = InMemorySkillStore() + await init_client(options={"skillStore": store}, client=object()) + with pytest.raises(ValueError, match="debounce"): + await watch_skills("*", tmp_path / "s", debounce=debounce) + # Refused before anything was attached, so there is no watcher to close. + assert store._listeners.get(SKILL_OBJECT_KIND, []) == [] + + async def test_on_reconcile_receives_each_subsequent_report( + self, tmp_path: Any, make_raw_skill: Any + ) -> None: + """Not the initial report: that one is returned to the caller directly.""" + store = InMemorySkillStore() + store.put(make_raw_skill(key="a", version=1, content="body")) + await init_client(options={"skillStore": store}, client=object()) + seen: list[Any] = [] + report, watcher = await watch_skills( + "*", tmp_path / "s", debounce=0.05, on_reconcile=seen.append + ) + try: + assert seen == [] + store.put(make_raw_skill(key="a", version=2, content="new body")) + assert wait_until(lambda: len(seen) >= 1, timeout=10) + assert seen[0] is not report + assert seen[0].ok is True + assert [a.key for a in seen[0].actions] == ["a"] + finally: + watcher.close() + + async def test_a_reconcile_that_raises_does_not_kill_the_watcher( + self, tmp_path: Any, make_raw_skill: Any, caplog: Any + ) -> None: + """A watcher that died on one bad run would silently stop pruning.""" + store = InMemorySkillStore() + await init_client(options={"skillStore": store}, client=object()) + calls: list[int] = [] + + def explodes_once(_report: Any) -> None: + calls.append(1) + if len(calls) == 1: + raise RuntimeError("boom") + + _report, watcher = await watch_skills( + "*", tmp_path / "s", debounce=0.05, on_reconcile=explodes_once + ) + try: + with caplog.at_level("ERROR"): + store.put(make_raw_skill(key="a", version=1, content="one")) + assert wait_until(lambda: len(calls) >= 1, timeout=10) + store.put(make_raw_skill(key="a", version=2, content="two")) + assert wait_until(lambda: len(calls) >= 2, timeout=10) + written = tmp_path / "s" / "a" / "SKILL.md" + assert wait_until(lambda: written.read_text() == "two", timeout=10) + finally: + watcher.close() + + async def test_every_write_skills_option_is_passed_straight_through( + self, tmp_path: Any, make_raw_skill: Any + ) -> None: + """``prune``, ``timeout`` and ``on_unavailable`` are not reinterpreted. + + Driven through one of ``write_skills``' own cases: a store that cannot + answer, with ``on_unavailable="raise"``, raises out of the initial + reconcile exactly as it would out of ``write_skills``. + """ + + class Unavailable(InMemorySkillStore): + def all_objects(self, _kind: str) -> dict[str, Any]: + raise RuntimeError("store is down") + + store = Unavailable() + await init_client(options={"skillStore": store}, client=object()) + with pytest.raises(RuntimeError, match="store is down"): + await watch_skills( + "*", tmp_path / "s", on_unavailable="raise", debounce=0.05 + ) + assert store._listeners.get(SKILL_OBJECT_KIND, []) == [] + + async def test_prune_is_passed_through_and_can_be_turned_off( + self, tmp_path: Any, make_raw_skill: Any + ) -> None: + store = InMemorySkillStore() + store.put(make_raw_skill(key="a", version=1, content="body")) + await init_client(options={"skillStore": store}, client=object()) + report, watcher = await watch_skills( + "*", tmp_path / "s", prune=False, debounce=0.05 + ) + watcher.close() + written = tmp_path / "s" / "a" / "SKILL.md" + assert written.read_text() == "body" + assert report.ok is True + assert all(a.action != "removed" for a in report.actions) + + async def test_an_invalid_root_raises_out_of_watch_skills( + self, tmp_path: Any + ) -> None: + """The initial reconcile runs on the caller's thread, so a bad root is + the caller's exception rather than a line in a worker thread's log.""" + store = InMemorySkillStore() + await init_client(options={"skillStore": store}, client=object()) + not_a_directory = tmp_path / "file" + not_a_directory.write_text("") + with pytest.raises(ValueError, match="not a directory"): + await watch_skills("*", not_a_directory, debounce=0.05) + async def test_a_store_with_no_listener_support_is_refused_loudly( self, tmp_path: Any ) -> None: