From 224f1ab877f98b42b2ede35f8d5eafd6523c4357 Mon Sep 17 00:00:00 2001 From: Christie Williams Date: Fri, 23 Oct 2026 16:46:47 -0400 Subject: [PATCH] fix(skills): coalesce a watcher's debounce window into one reconcile SkillWatcher cleared its wake flag before waiting out the debounce, so a notification arriving inside the window scheduled a second reconcile right after the first. A burst spread across one window ran two reconciles instead of one. The debounce now trails: after each window the watcher checks whether another change arrived and, if so, waits a fresh window. It reconciles only once delivery has been quiet for a whole window. The flag is still cleared before each window, so a change landing during the reconcile itself schedules one more pass. Co-Authored-By: Claude Opus 5.5 --- packages/client/README.md | 2 +- .../launchdarkly_ai_server/skills_watch.py | 36 +++++++++----- packages/client/tests/test_skills_watch.py | 48 +++++++++++++++++-- 3 files changed, 67 insertions(+), 19 deletions(-) diff --git a/packages/client/README.md b/packages/client/README.md index 924aefdf..fdbfa5cf 100644 --- a/packages/client/README.md +++ b/packages/client/README.md @@ -850,7 +850,7 @@ that skips verification. | `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. | | `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=…, 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. `close()` is **final** — `start()` afterwards raises. `poll_interval`, `read_timeout`, `initial_backoff` and `max_backoff` must be positive and finite, and `initial_backoff` may not exceed `max_backoff`. **Server-side only.** 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, so revocation takes effect within `debounce` rather than at the next restart. Returns `(initial report, SkillWatcher)`; close the watcher when done. `debounce` is in **seconds**, non-negative and finite. `on_reconcile` receives each *subsequent* report. One watcher per root. | +| `watch_skills(skills, root, *, debounce=0.5, on_reconcile=None, …)` | `write_skills` plus a re-reconcile on every delivery change, so revocation takes effect once delivery has been quiet for `debounce` rather than at the next restart. Returns `(initial report, SkillWatcher)`; close the watcher when done. `debounce` is in **seconds**, non-negative and finite. `on_reconcile` receives each *subsequent* report. One watcher per root. | | `StoreDiagnostics` | What the transport has seen: `payloads_transferred`, `skill_objects_received`, `objects_ignored`, `objects_revoked`, `payloads_ignored`, `hashless_objects`, `connection_failures`, `last_error`. | Configure the store with `set_skill_store(store)`. It applies on every call, before or after diff --git a/packages/client/src/launchdarkly_ai_server/skills_watch.py b/packages/client/src/launchdarkly_ai_server/skills_watch.py index 0283a2d3..4a65a956 100644 --- a/packages/client/src/launchdarkly_ai_server/skills_watch.py +++ b/packages/client/src/launchdarkly_ai_server/skills_watch.py @@ -3,10 +3,11 @@ ``write_skills`` is a one-shot reconcile of what the store holds now. ``watch_skills`` re-runs it whenever the store reports a change. With ``"*"``, -a skill revoked in LaunchDarkly is removed from disk within a debounce interval -rather than at the next restart. An explicit list is fixed: a skill the store -answers ``absent`` for keeps its files and reports an error, and a config change -that unpins a skill or moves it to a new version is not seen. +a skill revoked in LaunchDarkly is removed from disk once delivery has been +quiet for a debounce interval, rather than at the next restart. An explicit +list is fixed: a skill the store answers ``absent`` for keeps its files and +reports an error, and a config change that unpins a skill or moves it to a new +version is not seen. ``on_unavailable="keep"`` remains the default, so an outage never deletes the application's skill files. @@ -30,8 +31,9 @@ DEFAULT_DEBOUNCE_SECONDS = 0.5 """ -Default debounce, in seconds: how long to wait after a change before -reconciling, so a payload of many skills triggers one reconcile, not one each. +Default debounce, in seconds: how long delivery must stay quiet after a change +before reconciling, so a payload of many skills triggers one reconcile, not one +each. """ @@ -104,11 +106,18 @@ def _run(self) -> None: continue if self._stop.is_set(): return - # Clear before the debounce so a change arriving mid-debounce - # triggers another pass instead of being swallowed by this one. - self._wake.clear() - if self._stop.wait(self._debounce): - return + # A trailing debounce: wait until a whole window passes with no new + # change, so a burst spread across the window reconciles once. The + # reconcile reads the store after the last window, so it sees every + # change in the burst. Clearing before each window, not after the + # reconcile, means a change arriving during the reconcile itself + # still schedules one more pass instead of being swallowed by it. + while True: + self._wake.clear() + if self._stop.wait(self._debounce): + return + if not self._wake.is_set(): + break self._reconcile_once() def _reconcile_once(self) -> None: @@ -205,8 +214,9 @@ async def watch_skills( the store's lifetime. Args: - debounce: Seconds to wait after a change before reconciling. Must be a - non-negative finite number. + debounce: Seconds delivery must stay quiet after a change before + reconciling; each further change restarts the wait, so a burst + reconciles once. Must be a non-negative finite number. on_reconcile: Called with each re-reconcile's report (not the initial one). Exceptions it raises are logged and do not stop the watcher. diff --git a/packages/client/tests/test_skills_watch.py b/packages/client/tests/test_skills_watch.py index 72dee6d6..9f48adad 100644 --- a/packages/client/tests/test_skills_watch.py +++ b/packages/client/tests/test_skills_watch.py @@ -67,15 +67,15 @@ async def test_the_in_memory_store_can_also_drive_a_watch( 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. + """Six notifications spread across one window produce one reconcile. To discriminate rather than pass vacuously, the puts happen after the listener is attached, the counter is asserted against an exact figure, and each put lands in its own scheduler pass (a back-to-back burst would collapse to one reconcile with or without a debounce). - Two, not one: the wake flag is cleared before the window, so a change - arriving inside it schedules one further pass. + The debounce trails: each change restarts the window, so the reconcile + runs once delivery goes quiet and sees every skill in the burst. """ store = InMemorySkillStore() set_skill_store(store) @@ -85,10 +85,48 @@ async def test_a_burst_of_changes_coalesces_into_a_debounce_window( for i in range(6): store.put(make_raw_skill(key=f"skill-{i}")) 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. + assert wait_until(lambda: watcher.reconciles >= 1, timeout=10) + # Settle well past two windows, so a second would have landed by now. time.sleep(1.5) + assert watcher.reconciles == 1 + for i in range(6): + assert (tmp_path / "s" / f"skill-{i}" / "SKILL.md").exists() + finally: + watcher.close() + + async def test_a_change_during_a_reconcile_schedules_one_more( + self, tmp_path: Any, make_raw_skill: Any + ) -> None: + """The wait for quiet ends before the reconcile starts, so a change + landing while it runs is not swallowed by it: exactly one more pass + follows, and that pass writes the late skill.""" + + class ChangesDuringSnapshot(InMemorySkillStore): + """Delivers a new skill the moment an armed reconcile has taken its + snapshot, so the snapshot cannot include it.""" + + def __init__(self) -> None: + super().__init__() + self.armed = False + + def all_objects(self, kind: str) -> dict[str, dict[str, Any]]: + objects = super().all_objects(kind) + if self.armed: + self.armed = False + self.put(make_raw_skill(key="late")) + return objects + + store = ChangesDuringSnapshot() + set_skill_store(store) + _report, watcher = await watch_skills("*", tmp_path / "s", debounce=0.05) + try: + store.armed = True + store.put(make_raw_skill(key="early")) + assert wait_until(lambda: watcher.reconciles >= 2, timeout=10) + # Settle well past the window, so a third would have landed by now. + time.sleep(0.5) assert watcher.reconciles == 2 + assert (tmp_path / "s" / "late" / "SKILL.md").exists() finally: watcher.close()