httk.serve.dsp.state¶

Locked, non-durable state repository for DSP negotiations and transfers.

Classes¶

DspState

Own immutable DSP process snapshots behind one asynchronous lock.

Module Contents¶

class httk.serve.dsp.state.DspState¶

Own immutable DSP process snapshots behind one asynchronous lock.

The repository is deliberately in-memory only. Callback transitions are reserved with a unique pending token while holding the lock, sent by the provider outside the lock, then committed only when the reservation still belongs to that callback.

async create_negotiation(record)¶

Store a new negotiation process.

Parameters:: record (httk.serve.dsp.models.NegotiationRecord) – Initial negotiation record to store.
Returns:: Stored immutable snapshot.
Raises:: ValueError – If its provider process identifier already exists.
Return type:: httk.serve.dsp.models.NegotiationRecord

async create_transfer(record)¶

Store a new transfer process.

Parameters:: record (httk.serve.dsp.models.TransferRecord) – Initial transfer record to store.
Returns:: Stored immutable snapshot.
Raises:: ValueError – If its provider process identifier already exists.
Return type:: httk.serve.dsp.models.TransferRecord

async create_or_transfer_for_consumer(record)¶

Atomically create a transfer or recover one with the same consumer PID.

Parameters:: record (httk.serve.dsp.models.TransferRecord) – Candidate initial transfer record.
Returns:: (record, True) when stored, otherwise the existing record and False.
Raises:: ValueError – If its provider process identifier already exists.
Return type:: tuple[httk.serve.dsp.models.TransferRecord, bool]

async get_or_create_transfer_for_consumer(consumer_pid, create)¶

Atomically recover or construct one transfer for a consumer process ID.

create runs only when no existing consumer process is found and is called while the lock is held. It must perform no I/O; this lets callers defer ID allocation until an idempotent retry is known to be new.

Parameters:

consumer_pid (str) – Consumer transfer-process identifier to recover.
create (collections.abc.Callable[[], httk.serve.dsp.models.TransferRecord]) – Zero-argument constructor for a new immutable record.

Returns:

(record, True) when stored, otherwise the existing record and False.

Raises:

ValueError – If the constructed record conflicts or uses a different consumer PID.

Return type:

tuple[httk.serve.dsp.models.TransferRecord, bool]

async negotiation(provider_pid)¶

Return one negotiation snapshot.

Parameters:: provider_pid (str) – Provider process identifier to resolve.
Returns:: Immutable negotiation snapshot.
Raises:: KeyError – If no such process exists.
Return type:: httk.serve.dsp.models.NegotiationRecord

async transfer(provider_pid)¶

Return one transfer snapshot.

Parameters:: provider_pid (str) – Provider transfer-process identifier to resolve.
Returns:: Immutable transfer snapshot.
Raises:: KeyError – If no such process exists.
Return type:: httk.serve.dsp.models.TransferRecord

async transfer_for_consumer(consumer_pid)¶

Return an existing transfer identified by its consumer process ID.

Parameters:: consumer_pid (str) – Consumer transfer-process identifier to find.
Returns:: Matching immutable snapshot, or None when absent.
Return type:: httk.serve.dsp.models.TransferRecord | None

async finalized_agreement(agreement_id)¶

Return an agreement only if its negotiation has reached FINALIZED.

Parameters:: agreement_id (str) – Agreement identifier to resolve.
Returns:: Finalized agreement record, or None when unavailable.
Return type:: httk.serve.dsp.models.AgreementRecord | None

async agreement(agreement_id)¶

Return an agreement and its negotiation state regardless of finalization.

Parameters:: agreement_id (str) – Agreement identifier to resolve.
Returns:: Agreement record paired with its current negotiation state, or None when no process owns the identifier.
Return type:: tuple[httk.serve.dsp.models.AgreementRecord, str] | None

async reserve_negotiation(provider_pid, *, expected_states, transition)¶

Reserve an outbound negotiation transition without holding the lock for I/O.

Parameters:

provider_pid (str) – Provider process identifier to transition.
expected_states (frozenset[str]) – States from which the transition is valid.
transition (str) – Descriptive local transition name.

Returns:

Pre-transition snapshot and unique reservation token.

Raises:

KeyError – If no such process exists.
RuntimeError – If another callback is pending or the state is invalid.

Return type:

tuple[httk.serve.dsp.models.NegotiationRecord, str]

async reserve_transfer(provider_pid, *, expected_states, transition)¶

Reserve an outbound transfer transition without holding the lock for I/O.

Parameters:

provider_pid (str) – Provider transfer-process identifier to transition.
expected_states (frozenset[str]) – States from which the transition is valid.
transition (str) – Descriptive local transition name.

Returns:

Pre-transition snapshot and unique reservation token.

Raises:

KeyError – If no such process exists.
RuntimeError – If another callback is pending or the state is invalid.

Return type:

tuple[httk.serve.dsp.models.TransferRecord, str]

async commit_negotiation(provider_pid, token, *, state, agreement)¶

Commit an acknowledged negotiation callback when its token still matches.

Parameters:

provider_pid (str) – Provider process identifier to update.
token (str) – Reservation token returned by reserve_negotiation().
state (str) – Newly acknowledged DSP state.
agreement (httk.serve.dsp.models.AgreementRecord | None) – Agreement to store, if an agreement was acknowledged.

Returns:

Whether this reservation was still current and was committed.

Return type:

bool

async commit_transfer(provider_pid, token, *, state)¶

Commit an acknowledged transfer callback when its token still matches.

Parameters:

provider_pid (str) – Provider transfer-process identifier to update.
token (str) – Reservation token returned by reserve_transfer().
state (str) – Newly acknowledged DSP state.

Returns:

Whether this reservation was still current and was committed.

Return type:

bool

async fail_negotiation(provider_pid, token, *, detail, retries)¶

Record an unacknowledged negotiation callback failure.

Parameters:

provider_pid (str) – Provider process identifier to update.
token (str) – Reservation token returned by reserve_negotiation().
detail (str) – Safe callback failure detail.
retries (int) – Delivery attempts used for the failed callback.

Returns:

Whether this reservation was still current and was updated.

Return type:

bool

async fail_transfer(provider_pid, token, *, detail, retries)¶

Record an unacknowledged transfer callback failure.

Parameters:

provider_pid (str) – Provider transfer-process identifier to update.
token (str) – Reservation token returned by reserve_transfer().
detail (str) – Safe callback failure detail.
retries (int) – Delivery attempts used for the failed callback.

Returns:

Whether this reservation was still current and was updated.

Return type:

bool

async receive_negotiation(provider_pid, *, expected_states, state)¶

Commit an inbound negotiation transition atomically.

Parameters:

provider_pid (str) – Provider process identifier to update.
expected_states (frozenset[str]) – States from which the inbound transition is valid.
state (str) – Newly received DSP state.

Returns:

Updated immutable process snapshot.

Raises:

KeyError – If no such process exists.
RuntimeError – If another callback is pending or the state is invalid.

Return type:

httk.serve.dsp.models.NegotiationRecord

async receive_transfer(provider_pid, *, expected_states, state)¶

Commit an inbound transfer transition atomically.

Parameters:

provider_pid (str) – Provider transfer-process identifier to update.
expected_states (frozenset[str]) – States from which the inbound transition is valid.
state (str) – Newly received DSP state.

Returns:

Updated immutable process snapshot.

Raises:

KeyError – If no such process exists.
RuntimeError – If another callback is pending or the state is invalid.

Return type:

httk.serve.dsp.models.TransferRecord