s-replicationkit — доставка событий между узлами

Кит родился из волны 7 платформы китов (Atlas #2718) и закрывает четыре дыры синхронизации, найденные в коде Atlas и подтверждённые комментариями самого этого кода.

И пятое, ради чего всё затевалось: поток, не теряющий события на разрыве. Кадр несёт номер (id:), клиент возвращается с Last-Event-ID, сервер отдаёт хвост из stream.ReplayBuffer — а если хвост уже вытеснен, честно говорит об этом (truncated=True), а не молчит.

Состав

Зависимостей нет — только stdlib. Доставка стоит под всеми, и кит, который тянет за собой транспорт или ORM, навязывает их каждому потребителю.

Как это выглядит

Применение с честным курсором:

from replicationkit import FileCursorStore, apply_stream

report = apply_stream(envelopes, apply_one, cursor_store=FileCursorStore(path), channel="atlas")
if not report.complete:
    log.warning("отложено %s конвертов: %s", report.skipped, report.pending)

Отказ доставки, разобранный по существу:

from replicationkit import decide

record.attempts += 1
verdict = decide(exc, record.attempts, permanent=(AccessRevoked,))
record.status = verdict.status        # 'failed' сразу — или 'pending' до порога
record.last_error = verdict.reason

Поток, который переживает разрыв:

from replicationkit import Envelope, ReplayBuffer, SequenceAllocator

seq, buffer = SequenceAllocator(), ReplayBuffer(maxlen=512)
buffer.add(Envelope(seq=seq.next(), envelope_id=job_id, kind="job.status", payload=state))

tail, truncated = buffer.since(parse_cursor(request.headers.get("Last-Event-ID")))

Переезды и переходники

telemetrykit.outbox остался как переходник: его публичный API (append / read_batch / remove / path) не изменился, а механика теперь живёт в replicationkit.filequeue.FileEnvelopeQueue. Причина переезда: очередь исходящих конвертов — это не про телеметрию, через неё едет что угодно, и второй такой механизм заводить нельзя. Чтение переменных окружения с префиксом продукта осталось у переходника: кит доставки имени продукта не знает.

Образец удержания соединения (hold_channel) поднят из netkit.hold — там он написан и обкатан на бою; сам netkit не тронут.

Прогон

uv sync --group dev
uv run pytest -q
uv run ruff check .
uv run --with import-linter lint-imports