fix: keep stats state consistent across workers

routstr/core/ledger_analytics.py

@@ -4,7 +4,7 @@ from copy import deepcopy
 from datetime import UTC, datetime, timedelta
 from typing import Any
 
-from sqlalchemy import case
+from sqlalchemy import case, text
 from sqlmodel import col, func, select
 
 from . import terminal_outcomes
@@ -158,6 +158,13 @@ async def get_ledger_usage_dashboard(
         + col(TerminalOutcome.cache_creation_input_tokens)
     )
     async with session_factory() as session:
+        # Keep all aggregates on one snapshot while queued outcomes are flushed.
+        if session.get_bind().dialect.name == "sqlite":
+            await session.execute(text("BEGIN"))
+        else:
+            await session.connection(
+                execution_options={"isolation_level": "REPEATABLE READ"}
+            )
         bucket_rows = (
             await session.exec(
                 select(bucket, *measures)

routstr/nostr/analytics_runtime.py

@@ -4,14 +4,15 @@ import asyncio
 import time
 from datetime import UTC, datetime
 
-from ..core.db import create_session
+from ..core.db import NsecState, Secret, create_session
 from ..core.logging import get_logger
-from ..core.settings import SettingsService, settings
+from ..core.settings import SettingsService, derive_npub_from_nsec, settings
 from ..core.terminal_outcomes import (
     start_terminal_outcome_writer,
     stop_terminal_outcome_writer,
     terminal_outcome_writer,
 )
+from ..core.vault import decrypt
 from .analytics_v2_delivery import (
     AnalyticsV2Delivery,
     AnalyticsV2Producer,
@@ -34,6 +35,15 @@ async def _read_state() -> DeliveryStateSnapshot:
     state = await get_analytics_v2_delivery_state(create_session)
     async with create_session() as session:
         await SettingsService.refresh(session, ("enable_analytics_sharing",))
+        secret = await session.get(Secret, 1)
+        if secret is not None and secret.nsec_state != NsecState.legacy:
+            if secret.nsec_state == NsecState.encrypted:
+                if secret.encrypted_nsec is None:
+                    raise RuntimeError("Encrypted Nostr identity has no ciphertext")
+                settings.nsec = decrypt(secret.encrypted_nsec)
+            else:
+                settings.nsec = ""
+            settings.npub = derive_npub_from_nsec(settings.nsec) or ""
     return state
 
 

tests/unit/test_analytics_runtime.py

@@ -1,7 +1,9 @@
 from __future__ import annotations
 
 import asyncio
+import importlib
 import json
+import time
 from collections.abc import AsyncIterator
 from contextlib import asynccontextmanager
 from datetime import UTC, datetime, timedelta
@@ -10,14 +12,17 @@ from typing import Any
 
 import pytest
 import pytest_asyncio
+from httpx import ASGITransport, AsyncClient
 from sqlalchemy.ext.asyncio import create_async_engine
 from sqlmodel import SQLModel, col, select, text
 from sqlmodel.ext.asyncio.session import AsyncSession
 
-from routstr.core import terminal_outcomes
-from routstr.core.db import TerminalOutcomeEpoch
-from routstr.core.settings import SettingsService
+from routstr.core import admin, terminal_outcomes, vault
+from routstr.core.db import Secret, TerminalOutcomeEpoch
+from routstr.core.main import app
+from routstr.core.settings import Settings, SettingsService
 from routstr.nostr import analytics_runtime as runtime
+from routstr.nostr import listing
 
 
 @pytest_asyncio.fixture
@@ -165,6 +170,79 @@ async def test_identity_rotation_cancels_previous_publisher(
     assert node.writer.running
 
 
+@pytest.mark.asyncio
+async def test_saved_identity_reaches_other_workers_without_disabling_sharing(
+    node: Any, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    settings_module = importlib.import_module("routstr.core.settings")
+    first = runtime.settings.copy(deep=True)
+    first.nsec = ""
+    first.npub = ""
+    first.enable_analytics_sharing = True
+    second = first.copy(deep=True)
+
+    def bind_worker(settings: Settings) -> None:
+        for module in (runtime, admin, listing, settings_module):
+            monkeypatch.setattr(module, "settings", settings)
+        monkeypatch.setattr(SettingsService, "_current", settings)
+
+    async with node.sessions() as session:
+        await session.exec(  # type: ignore[call-overload]
+            text("INSERT INTO settings (id, data) VALUES (1, :data)").bindparams(
+                data=json.dumps(settings_module._strip_secret_fields(first.dict()))
+            )
+        )
+        await session.commit()
+    monkeypatch.setattr(admin, "create_session", node.sessions)
+    token = "stats-identity-test"
+    monkeypatch.setitem(admin.admin_sessions, token, int(time.time()) + 60)
+    other = runtime.AnalyticsCoordinator()
+    try:
+        for coordinator, settings in ((node.coordinator, first), (other, second)):
+            bind_worker(settings)
+            await coordinator.prepare_startup()
+            await coordinator.sync_once()
+        async with AsyncClient(
+            transport=ASGITransport(app=app),
+            base_url="http://test",
+            headers={"Authorization": f"Bearer {token}"},
+        ) as client:
+            for key in ("11" * 32, "22" * 32, ""):
+                bind_worker(first)
+                response = await client.patch("/admin/api/nsec", json={"nsec": key})
+                assert response.status_code == 200
+                saved = await client.patch(
+                    "/admin/api/settings", json={"npub": response.json()["npub"]}
+                )
+                assert saved.status_code == 200
+                async with node.sessions() as session:
+                    secret = await session.get(Secret, 1)
+                    assert secret is not None
+                    assert (
+                        vault.decrypt(secret.encrypted_nsec)
+                        if secret.encrypted_nsec
+                        else ""
+                    ) == key
+                await node.coordinator.sync_once()
+                state = await runtime.get_analytics_v2_delivery_state(node.sessions)
+                for coordinator, settings in (
+                    (other, second),
+                    (node.coordinator, first),
+                    (other, second),
+                ):
+                    bind_worker(settings)
+                    await coordinator.sync_once()
+                    current = await runtime.get_analytics_v2_delivery_state(
+                        node.sessions
+                    )
+                    assert current.sharing_enabled is bool(key)
+                    assert current.generation == state.generation
+                    assert settings.nsec == key
+                    assert settings.npub == response.json()["npub"]
+    finally:
+        await other.close()
+
+
 @pytest.mark.asyncio
 async def test_restart_writer_failure_disables_publication_before_serving(
     node: Any, monkeypatch: Any

tests/unit/test_ledger_analytics.py

@@ -5,10 +5,11 @@ from contextlib import asynccontextmanager
 from copy import deepcopy
 from datetime import UTC, date, datetime, timedelta
 from pathlib import Path
+from typing import Any
 
 import pytest
 from sqlalchemy.ext.asyncio import create_async_engine
-from sqlmodel import SQLModel
+from sqlmodel import SQLModel, text
 from sqlmodel.ext.asyncio.session import AsyncSession
 
 from routstr.core.db import (
@@ -225,6 +226,58 @@ async def test_dashboard_replaces_log_totals_and_matches_every_chart(
     }
 
 
+@pytest.mark.parametrize("new_bucket", [False, True])
+async def test_dashboard_keeps_one_snapshot_while_queued_outcomes_flush(
+    sessions: SessionFactory,
+    monkeypatch: pytest.MonkeyPatch,
+    new_bucket: bool,
+) -> None:
+    async with sessions() as session:
+        await session.execute(text("PRAGMA journal_mode=WAL"))
+    await _seed(sessions)
+    original_exec = AsyncSession.exec
+    flushed = False
+
+    async def exec_then_flush(
+        session: AsyncSession, statement: Any, **kwargs: Any
+    ) -> Any:
+        nonlocal flushed
+        result = await original_exec(session, statement, **kwargs)
+        if not flushed:
+            flushed = True
+            async with sessions() as writer:
+                writer.add(
+                    _row(
+                        "queued",
+                        NOW - timedelta(minutes=1 if new_bucket else 240),
+                    )
+                )
+                await writer.commit()
+        return result
+
+    monkeypatch.setattr(AsyncSession, "exec", exec_then_flush)
+    result = await get_ledger_usage_dashboard(
+        _legacy(), interval=60, hours=24, session_factory=sessions, now=NOW
+    )
+    assert flushed
+    assert result["summary"]["successful_chat_completions"] == 3
+    assert result["summary"]["revenue_msats"] == 1250
+    assert (
+        sum(model["successful"] for model in result["revenue_by_model"]["models"]) == 3
+    )
+    for point in result["model_usage_mix"]["metrics"]:
+        assert point["others"] >= 0
+        assert point["others_revenue_msats"] >= 0
+        assert (
+            sum(point["model_counts"].values()) + point["others"]
+            == point["total_successful"]
+        )
+        assert (
+            sum(point["model_revenue_msats"].values()) + point["others_revenue_msats"]
+            == point["total_revenue_msats"]
+        )
+
+
 async def test_historical_dates_do_not_return_recent_ledger_or_log_data(
     sessions: SessionFactory,
 ) -> None: