perf: Reuse the connection to Lago, and fix two silent-failure paths

[ancorcruz]

Context

Every batch opened a new connection. LagoClient.send_batch called the module-level requests.post, which builds a Session, uses it, and closes it — discarding the connection pool after every call. HTTP/1.1 is persistent by default and urllib3 pools connections, so keep-alive was available the whole time; the client was declining it. Every batch paid a TCP handshake plus a TLS handshake — about 2 RTT — in front of a request the server answers in ~147 ms. Discarding the Session discarded the TLS session ticket too, so not even an abbreviated handshake was possible on reconnect.

That cost is amortised over 100 events in the normal path. It is not amortised at all on the recovery path: when one duplicate transaction_id makes Lago reject a whole batch, _send_individually re-sends all 100 events one at a time, and the handshake was paid per event — roughly a hundred times the per-event overhead, precisely when the system is already degraded.

flush() reported success on events it hadn't delivered. _take_batch pops events out of the buffer before attempting the send, and flush() decided it was done by checking that the buffer was empty. For the duration of a POST the batch sat in neither the buffer nor Lago, so flush() returned True on events that were still in flight — and, if that send then failed, on events that had not arrived at all. It is the documented delivery checkpoint, so a caller using it as one (end of a batch-job phase, before logging "billing synced", before a container reports ready-to-terminate) got a false positive with nothing behind it.

A provider changing its response shape produced silence. LagoSDK.emit() reports failures through on_error, but the adapter that parses a provider response runs outside it, on the wrapper's side of the boundary. An adapter that raised did so before emit() was ever entered, so the SDK's own reporting never ran — across the five wrappers, 14 except Exception sites, 14 logger.warning calls, and zero routes to on_error. Provider drift doesn't hit an occasional call; it hits every call for that provider. The result was a total billing outage that surfaced only as a log line.

Underneath all three, the test suite couldn't have caught them regressing: it stubbed HTTP by patching library functions, so a stub silently stops matching the moment the code under test changes call path, and the request goes out to the real network instead.

---

Description

Impact

	Before	After
TCP connections per 8 batches	8	1
Drain rate, 137 ms from the API	227 evt/s	677 evt/s
Handshake cost of a rejected-batch recovery	~27.7 s	~0.3 s
flush() with a batch in flight	returns True	waits
Provider drift on a wrapped client	log only	on_error fires
Socket after sdk.shutdown()	leaked	released

Only the two throughput rows are path-dependent — the saving is two round trips per batch, so it shrinks for a client sitting next to the API and grows for one far from it. The connection count and the three behaviour rows hold everywhere.

Connection reuse

LagoClient now holds one requests.Session for its lifetime. Measured against a local server answering Connection: keep-alive, 8 batches cost 1 connection instead of 8; against api.getlago.com from 137 ms away, the median batch went from 440.6 ms to 147.6 ms. On the recovery path, a rejected batch that fans out into 100 isolated sends went from 101 connections to 1.

The 2.98× figure isn't claimed as a constant, which is why the tests assert on connection counts rather than on timing.

One consequence points the other way and is worth knowing: that fan-out now reaches Lago about 3× faster, with no application-level rate limit in front of it. That argues for bounding the fan-out separately, not against this change.

Holding a Session also creates a socket that can outlive the SDK — requests.post closed its own after every call, so nothing lingered before. LagoSDK.shutdown() now releases it, after the queue's exit drain has finished sending through it. A send after shutdown still works, since a closed Session rebuilds its pool.

A truthful flush()

The queue now tracks events that have been taken off the buffer but not yet accounted for, and flush() waits on that as well as on the buffer. A True now means delivered, dropped for good, or back on the buffer — never "somewhere in a socket."

This is the part worth reviewing closely. Every path that consumes a batch has to settle it exactly once:

Path	Settled by
Delivered	_drain_buffer, after the send
Transient failure → re-queued	_replay_failed, in the same lock acquisition
Permanent failure → isolated	_send_individually, for what it resolved rather than re-queued
Exit drain, delivered or lost	_run

A missed settle leaks the counter upward and flush() never returns True again; a double settle reopens the original bug. There's a test for each direction.

Worth stating for anyone benchmarking the SDK: a True from flush() was never proof of delivery. Showing that a load test lost nothing means counting the events Lago accepted.

Adapter failures reach on_error

All 14 wrapper sites now report as well as log. _report_error already swallows anything a customer's callback raises, so the LLM call still returns normally either way — the change is only that a billing gap now appears on the channel customers actually watch, the same one the queue already uses for buffer overflow.

Each site was verified statically to sit inside an except ... as exc handler, pass the bound exception, and use the sdk reference genuinely in scope (self._sdk inside the anthropic stream managers, sdk elsewhere).

Tests that can't quietly stop testing

pytest-socket's --disable-socket is on by default: a unit test that opens a socket now fails saying so, instead of reaching the network and failing later for an unrelated-looking reason. On the two verify_ssl tests that had started issuing live requests, it turns a confusing LagoApiError: 404 after 0.97 s into SocketBlockedError after 0.15 s. It also removes the class of CI flake where a green suite depends on the network, and makes it impossible for a test run to reach production Lago. 127.0.0.1 stays allowed so a test can stand up a real server when the property under test is socket behaviour.

test_connection_reuse.py does exactly that — it counts accepts on a local HTTP/1.1 server rather than asserting that some function was called. Its three cases cover the three decisions made here: that a Session is held, that the recovery path shares it, and that shutdown releases it. All three fail against the previous code.

There is deliberately no test for connection recovery. urllib3 reconnects transparently after a clean close — measured, both Connection: close and an idle hang-up are handled with zero errors — and the only part that belongs to this SDK, the queue treating ConnectionError as transient, is already covered. Testing it here would test the HTTP library instead.

responses is added for stubbing at the requests transport boundary rather than by patching internals; existing tests are untouched, it's there for new ones. No cassette or auto-record layer — fixtures stay explicit, captured by a developer running tests/unit/adapters/fixtures/capture_*.py on purpose and reviewed in the diff.

The recovery path was mutation-tested: six deliberate breakages — isolation disabled, transient sub-failures dropped, FIFO inverted, on_error per event, backoff skipped on partial requeue, in-flight counter unsettled — all six caught.

608 tests · 91.56% coverage · ruff, format and mypy clean. No public API changes.

Measurement caveat: connection counts are localhost (pool mechanics) plus a single real path to api.getlago.com (RTT). Neither covers server-side idle timeouts or load-balancer connection limits.