Metadata-Version: 2.4
Name: natsio-counters
Version: 0.1.0
Summary: natsio extension: distributed counters (ADR-49) over JetStream counter streams
Keywords: nats,nats.io,jetstream,counters,distributed-counter,adr-49
Author: Corrupt Mane
Author-email: Corrupt Mane <corruptmane@gmail.com>
License-Expression: Apache-2.0
Classifier: Development Status :: 3 - Alpha
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Dist: natsio
Requires-Python: >=3.13
Project-URL: Homepage, https://github.com/corruptmane/natsio
Project-URL: Repository, https://github.com/corruptmane/natsio
Project-URL: Changelog, https://github.com/corruptmane/natsio/blob/main/extensions/natsio-counters/CHANGELOG.md
Description-Content-Type: text/markdown

# natsio-counters

Distributed counters over JetStream counter streams
([ADR-49](https://github.com/nats-io/nats-architecture-and-design/blob/main/adr/ADR-49.md)),
mirroring [`orbit.go/counters`](https://github.com/synadia-io/orbit.go/tree/main/counters).
Distribution `natsio-counters`, imported as `natsio.counters`. Zero runtime
dependencies beyond `natsio`. Pre-1.0, no API-stability promises.

Requires **nats-server 2.12+** (the `allow_msg_counter` stream feature).

## What it is

A *counter stream* is a normal JetStream stream created with
`allow_msg_counter`. Every subject the stream captures becomes an independent,
**arbitrary-precision** counter. Incrementing is an ordinary publish carrying a
`Nats-Incr` header; the server folds the delta into a running total, stores it
as the subject's latest message, and returns the new value in the PubAck — so a
bare `add` needs no follow-up read.

Python `int` is itself unbounded, so it is used directly for both deltas and
values — no big-integer wrapper.

## Usage

```python
import natsio
from natsio.counters import CounterConfig, create_counter, get_counter

nc = await natsio.connect("nats://localhost")
js = nc.jetstream()

# Create a counter stream (allow_msg_counter + allow_direct are forced on).
counter = await create_counter(js, CounterConfig(name="COUNTS", subjects=["events.>"]))
# ...or bind an existing one:  counter = await get_counter(js, "COUNTS")

# Increment / decrement — returns the new running total, straight from the PubAck.
await counter.add("events.orders", 1)      # -> 1
await counter.add("events.orders", 10)     # -> 11
await counter.add("events.orders", -4)     # negative deltas are fine  -> 7
await counter.add("events.orders", 0)      # zero is a valid no-op increment

# Read a single counter's current value.
value = await counter.load("events.orders")          # -> 7

# Full entry: value + most-recent increment + source history (aggregation only).
entry = await counter.get("events.orders")
print(entry.value, entry.incr, entry.sources)

# Enumerate many counters in one batch Direct Get (wildcards allowed).
async for entry in counter.get_multiple(["events.>"]):
    print(entry.subject, entry.value)
```

## API

| Symbol | Purpose |
|---|---|
| `create_counter(js, config)` | Create a counter stream and return its `Counter`. |
| `get_counter(js, name)` | Bind an existing counter stream by name. |
| `counter_from_stream(js, stream)` | Wrap an already-fetched counter `Stream` (no round-trip). |
| `Counter.add(subject, delta) -> int` | Increment/decrement; returns the new total. |
| `Counter.load(subject) -> int` | Current value of one subject's counter. |
| `Counter.get(subject) -> CounterEntry` | Value + last increment + sources. |
| `Counter.get_multiple(subjects) -> AsyncIterator[CounterEntry]` | Batch/wildcard enumeration. |
| `CounterConfig` | Counter-stream configuration. |
| `CounterEntry` | `subject`, `value`, `sources`, `incr`. |

Errors (all subclass `natsio.jetstream.JetStreamError`): `CounterNotEnabledError`,
`DirectAccessRequiredError`, `CounterNotFoundError`,
`CounterSubjectNotInitializedError`, `InvalidCounterValueError`.

## ADR-49 wire contract

Pinned by the wire-contract tests and exported as constants:

- **Increment header** `Nats-Incr` (`COUNTER_INCREMENT_HEADER`) — a base-10
  integer of any size, optional leading sign. A counter stream **rejects any
  publish without it** (`message counter increment is missing`).
- **Sources header** `Nats-Counter-Sources` (`COUNTER_SOURCES_HEADER`) —
  server-set JSON `{source_stream: {subject: contribution}}`; present only on
  aggregating counters.
- **Stored payload** `{"val": "<int-as-string>"}` — the value is a *string* so
  it survives languages whose JSON numbers cap at 2^53.
- **PubAck** carries the new total in its `val` field (fast feedback, no Get).

Notes: `allow_msg_counter` can be set **only at stream creation** (it is read-only
afterwards) and is **incompatible with per-message TTLs and message schedules**,
so `CounterConfig` intentionally does not expose those knobs. Reads use Direct
Get, so `allow_direct` is required (and forced on by `create_counter`).
