refactor(usage): address round-2 review on PR #871

- remove accidentally-committed .claude/scheduled_tasks.lock (Claude Code
  runtime artifact swept in by `git add -A`) and gitignore it; the rest
  of .claude/ stays tracked.
- store: cache UsageEventStore.shared() as a process-wide instance so the
  hot search path doesn't allocate a fresh wrapper per metered query (the
  wrapper is stateless beyond its storage handle).
- hooks: pass enabled=True directly (the outer guard already confirmed
  the flag) instead of re-reading settings.usage_metering_enabled.
- migration: document the no-TTL retention design (control-plane rollup
  owns the lifecycle; the data plane only appends).
- tests: assert the best-effort error path logs at WARNING (observability
  contract).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

.claude/scheduled_tasks.lock

@@ -1 +0,0 @@
-{"sessionId":"fdfa73f5-8734-459a-8f23-67e1eeaa804c","pid":13329,"procStart":"23713","acquiredAt":1780838157352}

.gitignore

@@ -10,6 +10,9 @@ __pycache__/
 settings.toml
 settings.local.toml
 
+# Claude Code runtime artifacts (the rest of .claude/ is tracked)
+.claude/scheduled_tasks.lock
+
 # Git
 worktrees/
 

nextcloud_mcp_server/alembic/versions/20260610_1200_007_add_usage_events.py

@@ -34,6 +34,9 @@ depends_on = None
 def upgrade() -> None:
     is_pg = op.get_bind().dialect.name == "postgresql"
 
+    # Retention: this table has no TTL by design — the control-plane rollup
+    # owns the lifecycle (it pulls rows read-only into usage_daily, then
+    # prunes once a day is reconciled). The data plane only appends.
     op.create_table(
         "usage_events",
         # Pod-generated idempotency key. UUID on Postgres; TEXT on SQLite,

nextcloud_mcp_server/server/semantic.py

@@ -534,9 +534,11 @@ def configure_semantic_tools(mcp: FastMCP):
                             "fusion": fusion,
                             "doc_types": doc_types,
                         },
-                        # Pass the already-resolved flag so the store doesn't
+                        # The outer guard already confirmed the flag, so pass
-                        # rebuild Settings on this hot query path (ADR-024).
+                        # enabled=True directly — the store then skips a second
-                        enabled=settings.usage_metering_enabled,
+                        # uncached Settings build on this hot query path
+                        # (ADR-024).
+                        enabled=True,
                     )
                 except Exception:
                     # Reached only when shared()/store construction itself

nextcloud_mcp_server/usage/store.py

@@ -51,18 +51,27 @@ _INSERT_SQL = (
 class UsageEventStore:
     """Append-only writer for the app-DB ``usage_events`` table."""
 
+    # Process-wide cached instance returned by ``shared()`` so the hot search
+    # path doesn't allocate a fresh wrapper per metered query. The store is
+    # stateless beyond its storage handle, so one instance is reusable.
+    _shared_instance: "UsageEventStore | None" = None
+
     def __init__(self, storage: RefreshTokenStorage) -> None:
         self._storage = storage
 
     @classmethod
     async def shared(cls) -> "UsageEventStore":
-        """Build a store backed by the process-wide storage singleton.
+        """Return the process-wide store backed by the storage singleton.
 
-        ``get_shared_storage()`` runs ``initialize()`` (and thus Alembic
+        Cached after first build: ``get_shared_storage()`` already returns the
-        migrations) on first access, so the ``usage_events`` table is present
+        cached :class:`RefreshTokenStorage` (running ``initialize()`` / Alembic
-        by the time any event is recorded.
+        on first access, so ``usage_events`` exists), and the wrapper itself is
+        stateless, so reusing one instance avoids a per-call allocation on the
+        ``nc_semantic_search`` hot path.
         """
-        return cls(await get_shared_storage())
+        if cls._shared_instance is None:
+            cls._shared_instance = cls(await get_shared_storage())
+        return cls._shared_instance
 
     async def record_usage_event(
         self,

nextcloud_mcp_server/vector/processor.py

@@ -839,9 +839,10 @@ async def _index_document(
                             "user_id": doc_task.user_id,
                             "total_chars": total_chars,
                         },
-                        # Pass the already-resolved flag so the store doesn't
+                        # The outer guard already confirmed the flag, so pass
-                        # rebuild Settings here (ADR-024).
+                        # enabled=True directly — the store then skips a second
-                        enabled=settings.usage_metering_enabled,
+                        # uncached Settings build here (ADR-024).
+                        enabled=True,
                     )
                 except Exception:
                     # Reached only when shared()/store construction itself

tests/unit/test_usage_store.py

@@ -12,6 +12,7 @@ that a DB failure is swallowed instead of surfacing to the caller.
 """
 
 import json
+import logging
 import tempfile
 import uuid
 from datetime import datetime, timezone
@@ -199,7 +200,7 @@ async def test_metadata_none_is_null(storage, monkeypatch):
     assert row[4] is None
 
 
-async def test_best_effort_swallows_db_errors(storage, monkeypatch):
+async def test_best_effort_swallows_db_errors(storage, monkeypatch, caplog):
     """A DB failure is logged + dropped, never raised into the caller."""
     _set_metering(monkeypatch, True)
     store = UsageEventStore(storage)
@@ -218,10 +219,16 @@ async def test_best_effort_swallows_db_errors(storage, monkeypatch):
     monkeypatch.setattr(storage, "acquire", _boom)
 
     # Must not raise.
+    with caplog.at_level(logging.WARNING, logger="nextcloud_mcp_server.usage.store"):
         await store.record_usage_event(metric="pages_chunks", value=1)
 
     assert recorded, "record_db_operation should be called on the error path"
     assert recorded[-1][3] == "error"
+    # The observability contract: the dropped write surfaces at WARNING.
+    assert any(
+        r.levelno == logging.WARNING and "usage metering write dropped" in r.message
+        for r in caplog.records
+    )
 
 
 async def test_best_effort_swallows_unserializable_metadata(storage, monkeypatch):