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[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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)[source]¶

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