fix(examples): reconnect the otel exporters after a dropped connection

CHANGELOG.md

@@ -30,6 +30,14 @@ there was nothing to have changed from. Everything above it is ordinary.
   does. Its maintainer processes were still opening the logs when the run
   ended, and cost it several seconds: `demo.main` took 3–11 s with them and
   about 1 s without.
+- **The OTel exporters reconnect after a dropped connection.** They used to
+  hold one publication for the life of the process, so a broker restart
+  ended a service's telemetry for good, silently. Each exporter now takes a
+  `common.Publisher` with the stream's URI: it connects on the first batch,
+  drops a connection that fails, and the next batch connects again. A
+  failed batch is still dropped rather than raised, so telemetry never fails
+  the application. One line is printed when telemetry starts being dropped
+  and one when it is published again.
 
 ## 0.13.1 — 2026-10-03
 

examples/otel/common.py

@@ -11,11 +11,11 @@
 import math
 from typing import TYPE_CHECKING, Any
 
+import streamcast
+
 if TYPE_CHECKING:
     from collections.abc import Sequence
 
-    import streamcast
-
 # -- the schema pieces --------------------------------------------------------------
 
 ANY_VALUE: dict[str, Any] = {
@@ -119,27 +119,73 @@ def scope(value: Any) -> dict[str, object] | None:  # noqa: ANN401 — Instrumen
 # -- publishing from OTel's export thread -----------------------------------------
 
 
-def publish(
-    publication: streamcast.Publication,
-    loop: asyncio.AbstractEventLoop,
-    rows: Sequence[dict[str, object]],
-) -> bool:
-    """Publish `rows` from an OTel exporter; True once the stream has them.
+class Publisher:
+    """One stream's publisher for OTel's exporters, reconnecting after a drop.
 
     OTel's batch processors call `export` on their own worker thread, and a
-    streamcast publisher lives on an event loop — so the batch is handed to
+    streamcast publisher lives on an event loop — so each batch is handed to
     the loop and the thread waits for the stream's acknowledgement. That makes
     a flush mean the rows are DURABLE, not merely sent.
+
+    It connects on the first batch, and a batch that fails on the connection
+    drops it, so the next batch connects again: a broker restart costs the
+    batches sent while it was down, not every batch after. The SDK calls
+    `export` every half second or so, which is all the retrying there is.
     """
-    if loop.is_closed():
-        # A provider flushing at interpreter exit, after the loop has gone:
-        # nothing can be published, and the SDK wants a result, not a raise.
-        return False
-
-    sent = asyncio.run_coroutine_threadsafe(publication.send_many(rows), loop)
-    try:
-        sent.result(timeout=10)
-    except Exception:  # noqa: BLE001 — the SDK wants a result, not an exception
-        return False
-
-    return True
+
+    def __init__(self, uri: str, loop: asyncio.AbstractEventLoop) -> None:
+        self.uri = uri
+        self._loop = loop
+        self._publication: streamcast.Publication | None = None
+        self._failing = False
+        # Two services' processors can export at once; one connection each time.
+        self._connecting = asyncio.Lock()
+
+    async def _send(self, rows: Sequence[dict[str, object]]) -> None:
+        async with self._connecting:
+            if self._publication is None:
+                self._publication = await streamcast.publish(self.uri)
+
+            publication = self._publication
+
+        try:
+            await publication.send_many(rows)
+        except streamcast.Rejected:
+            raise  # a bad row, on a connection that is fine
+
+        except BaseException:
+            if self._publication is publication:
+                self._publication = None  # the next batch reconnects
+
+            raise
+
+    def send(self, rows: Sequence[dict[str, object]]) -> bool:
+        """Publish `rows`; True once the stream has them. Never raises: failing
+        to send telemetry must not fail the application that made it."""
+        if self._loop.is_closed():
+            # A provider flushing at interpreter exit, after the loop has gone.
+            return False
+
+        sent = asyncio.run_coroutine_threadsafe(self._send(rows), self._loop)
+        try:
+            sent.result(timeout=10)
+        except Exception as exc:  # noqa: BLE001 — the SDK wants a result, not an exception
+            sent.cancel()
+            # Printed, not logged: a `logging` record would be exported
+            # through this same failing publisher. Once per outage.
+            if not self._failing:
+                print(f"telemetry to {self.uri} is being dropped: {exc!r}", flush=True)
+
+            self._failing = True
+            return False
+
+        if self._failing:
+            print(f"telemetry to {self.uri} is being published again", flush=True)
+
+        self._failing = False
+        return True
+
+    async def aclose(self) -> None:
+        if self._publication is not None:
+            await self._publication.close()
+            self._publication = None

examples/otel/logs.py

@@ -22,13 +22,10 @@
 from examples.otel import common
 
 if TYPE_CHECKING:
-    import asyncio
     from collections.abc import Sequence
 
     from opentelemetry.sdk._logs import ReadableLogRecord
 
-    import streamcast
-
 SCHEMA: dict[str, Any] = {
     "type": "object",
     "properties": {
@@ -79,15 +76,12 @@ def row(record: ReadableLogRecord) -> dict[str, object]:
 class StreamLogExporter(LogRecordExporter):
     """Publishes each batch of OTel log records to a stream."""
 
-    def __init__(
-        self, publication: streamcast.Publication, loop: asyncio.AbstractEventLoop
-    ) -> None:
-        self._publication = publication
-        self._loop = loop
+    def __init__(self, publisher: common.Publisher) -> None:
+        self._publisher = publisher
 
     def export(self, batch: Sequence[ReadableLogRecord]) -> LogRecordExportResult:
         rows = [row(record) for record in batch]
-        if common.publish(self._publication, self._loop, rows):
+        if self._publisher.send(rows):
             return LogRecordExportResult.SUCCESS
 
         return LogRecordExportResult.FAILURE

examples/otel/metrics.py

@@ -46,13 +46,9 @@
 from examples.otel import common
 
 if TYPE_CHECKING:
-    import asyncio
-
     from opentelemetry.sdk.metrics import Exemplar
     from opentelemetry.sdk.metrics.export import MetricsData
 
-    import streamcast
-
 
 def _list_of(items: dict[str, Any]) -> dict[str, Any]:
     return {"type": ["array", "null"], "items": items}
@@ -252,15 +248,13 @@ class StreamMetricExporter(MetricExporter):
 
     def __init__(
         self,
-        publication: streamcast.Publication,
-        loop: asyncio.AbstractEventLoop,
+        publisher: common.Publisher,
         preferred_temporality: dict[type, AggregationTemporality] | None = None,
     ) -> None:
         super().__init__(
             preferred_temporality=preferred_temporality or PREFERRED_TEMPORALITY
         )
-        self._publication = publication
-        self._loop = loop
+        self._publisher = publisher
 
     def export(
         self,
@@ -269,7 +263,7 @@ def export(
         **kwargs: object,  # noqa: ARG002
     ) -> MetricExportResult:
         batch = rows(metrics_data)
-        if not batch or common.publish(self._publication, self._loop, batch):
+        if not batch or self._publisher.send(batch):
             return MetricExportResult.SUCCESS
 
         return MetricExportResult.FAILURE

examples/otel/services.py

@@ -36,8 +36,7 @@
 from opentelemetry.sdk.trace.export import BatchSpanProcessor
 from opentelemetry.trace import SpanKind, Status, StatusCode
 
-import streamcast
-from examples.otel import logs, metrics, spans
+from examples.otel import common, logs, metrics, spans
 
 if TYPE_CHECKING:
     from opentelemetry.metrics import Counter, Histogram
@@ -248,16 +247,19 @@ def orders_forever(exporters: Exporters, stop: threading.Event) -> None:
 async def exporting(base: str):  # noqa: ANN201
     """A publisher on each stream, wrapped as the exporters OTel's SDK calls."""
     loop = asyncio.get_running_loop()
-    async with (
-        streamcast.publish(f"{base}/logs") as log_publication,
-        streamcast.publish(f"{base}/spans") as span_publication,
-        streamcast.publish(f"{base}/metrics") as metric_publication,
-    ):
+    publishers = [
+        common.Publisher(f"{base}/{name}", loop)
+        for name in ("logs", "spans", "metrics")
+    ]
+    try:
         yield Exporters(
-            logs.StreamLogExporter(log_publication, loop),
-            spans.StreamSpanExporter(span_publication, loop),
-            metrics.StreamMetricExporter(metric_publication, loop),
+            logs.StreamLogExporter(publishers[0]),
+            spans.StreamSpanExporter(publishers[1]),
+            metrics.StreamMetricExporter(publishers[2]),
         )
+    finally:
+        for publisher in publishers:
+            await publisher.aclose()
 
 
 async def run(broker: str) -> None:

examples/otel/spans.py

@@ -20,13 +20,10 @@
 from examples.otel import common
 
 if TYPE_CHECKING:
-    import asyncio
     from collections.abc import Sequence
 
     from opentelemetry.sdk.trace import ReadableSpan
 
-    import streamcast
-
 
 def _struct(properties: dict[str, Any]) -> dict[str, Any]:
     return {
@@ -137,15 +134,12 @@ def row(span: ReadableSpan) -> dict[str, object]:
 class StreamSpanExporter(SpanExporter):
     """Publishes each batch of finished OTel spans to a stream."""
 
-    def __init__(
-        self, publication: streamcast.Publication, loop: asyncio.AbstractEventLoop
-    ) -> None:
-        self._publication = publication
-        self._loop = loop
+    def __init__(self, publisher: common.Publisher) -> None:
+        self._publisher = publisher
 
     def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
         rows = [row(span) for span in spans]
-        if common.publish(self._publication, self._loop, rows):
+        if self._publisher.send(rows):
             return SpanExportResult.SUCCESS
 
         return SpanExportResult.FAILURE

tests/test_otel_example.py

@@ -9,10 +9,12 @@
 
 from __future__ import annotations
 
+import asyncio
 import math
 import re
 from typing import Any
 
+import litelink
 import pytest
 
 pytest.importorskip("opentelemetry.sdk", reason="the OTel example's dev dependency")
@@ -134,6 +136,57 @@ def test_each_kind(self, value, stored):
         assert common.any_value(value) == stored
 
 
+class TestThePublisher:
+    """What every exporter publishes through: telemetry must never fail the app."""
+
+    async def test_a_broker_restart_costs_only_the_batches_sent_while_it_was_down(
+        self, tmp_path, capsys
+    ):
+        schema = {
+            "type": "object",
+            "properties": {"i": {"type": "integer"}},
+            "required": ["i"],
+        }
+
+        async def broker(port: int = 0):  # noqa: ANN202
+            stream = streamcast.Stream.new("t", root=tmp_path, schema=schema)
+            return await streamcast.serve(
+                stream, "127.0.0.1", port, publish=True, maintain=False
+            )
+
+        server = await broker()
+        port = server.sockets[0].getsockname()[1]
+        publisher = common.Publisher(
+            f"ws://127.0.0.1:{port}/t", asyncio.get_running_loop()
+        )
+        # `send` blocks on the loop, as OTel's export thread does.
+        assert await asyncio.to_thread(publisher.send, [{"i": 1}])
+
+        server.close()
+        await server.wait_closed()
+        # Down: dropped, not raised. Twice, but reported once.
+        assert not await asyncio.to_thread(publisher.send, [{"i": 2}])
+        assert not await asyncio.to_thread(publisher.send, [{"i": 3}])
+
+        server = await broker(port)
+        try:
+            assert await asyncio.to_thread(publisher.send, [{"i": 4}])
+            await publisher.aclose()
+        finally:
+            server.close()
+            await server.wait_closed()
+
+        # One line when the outage starts, one when it ends: not one a batch.
+        printed = capsys.readouterr().out.splitlines()
+        assert len(printed) == 2
+        assert "is being dropped" in printed[0]
+        assert "is being published again" in printed[1]
+        with litelink.open(tmp_path, "t", read_only=True) as log:
+            stored = log.sql("SELECT i FROM log ORDER BY litelink_offset").read_all()
+
+        assert stored.column("i").to_pylist() == [1, 4]
+
+
 def emitted() -> list:
     """Real SDK log records: `logging` through OTel's handler, captured in memory."""
     import logging
@@ -482,7 +535,7 @@ def test_the_exporter_asks_for_delta_counters_and_histograms(self):
         from opentelemetry.sdk.metrics import Counter, Histogram, UpDownCounter
         from opentelemetry.sdk.metrics.export import AggregationTemporality
 
-        exporter = metrics.StreamMetricExporter(None, None)  # ty: ignore[invalid-argument-type]
+        exporter = metrics.StreamMetricExporter(None)  # ty: ignore[invalid-argument-type]
         preferred = exporter._preferred_temporality  # noqa: SLF001
         assert preferred is not None
         assert preferred[Counter] == AggregationTemporality.DELTA