pamoja.sync

Idiomatic store-and-forward facade.

The queue a node writes into while it has nowhere to send. An in-memory buffer suits a test or a process that will not outlive it; a file-backed one survives a reboot, which is what a node somewhere without reliable power actually needs.

A full store refuses the next append rather than dropping anything, so a record is never lost without the caller being told.

 1"""Idiomatic store-and-forward facade.
 2
 3The queue a node writes into while it has nowhere to send. An in-memory buffer
 4suits a test or a process that will not outlive it; a file-backed one survives a
 5reboot, which is what a node somewhere without reliable power actually needs.
 6
 7A full store refuses the next append rather than dropping anything, so a record
 8is never lost without the caller being told.
 9"""
10
11from __future__ import annotations
12
13from pamoja._native import Store
14
15__all__ = ["Store"]
class Store:

A store-and-forward buffer.

Handing a store to a ladder consumes it, because the ladder owns it from then on. A consumed store is emptied rather than left aliasing what now belongs to the ladder, so using one twice raises.

def memory(capacity=0):

Creates a buffer held in memory.

A full store refuses the next append rather than dropping anything, so a record is never lost without the caller being told.

def file(dir, capacity=0):

Opens a buffer backed by a directory, so it survives a restart.

A record a power cut interrupted mid-write is never seen, and one written before the cut is found again when the directory is reopened. capacity is the most records to hold, or 0 for no bound; a full store refuses the next append, which keeps a long outage from filling the disk.

def drain_to(self, /, transport, topic):

Drains the buffer onto a transport, publishing each record to topic, oldest first, and returns how many went out.

Each record leaves the buffer only once the transport has taken it, so a send that fails raises the transport's error and leaves that record and every one after it buffered, in order, for the next drain. The transport is driven, not consumed.

def append(self, /, record):

Adds a record to the end of the buffer.

def peek(self, /):

Reads the oldest record without removing it, or None when empty.

def pop(self, /):

Removes and returns the oldest record, or None when empty.

def peek_text(self, /):

Reads the oldest record as text without removing it, or None when empty.

Raises ValueError if the record is not UTF-8 text.

def pop_text(self, /):

Removes and returns the oldest record as text, or None when empty.

Raises ValueError if the record is not UTF-8 text.

def len(self, /):

How many records the buffer holds.

is_available

Whether this store is still holdable, or has been given to a ladder.