scalus.cardano.node.stream

Block stream event.

Attributes

Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Which blocks a subscriber wants to observe.

Simpler than the other two algebras on purpose: most subscribers want every block, or every block in a range. Content-based selection belongs on TransactionQuery, where it can be pushed down.

Attributes

Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Future-based streaming provider, mirroring BlockchainProvider.

This is the layer the in-repo implementations extend: Emulator and BlockfrostProvider are already BlockchainProviders, so F is fixed to Future while C stays open — a caller gets fs2, ox or pekko streams from them by having the right given in scope, and gets a working stream with no adapter at all by using ScalusAsyncSource.

Attributes

Supertypes: trait BlockchainProvider

trait BlockchainStreamReader

trait BlockchainReader

trait BlockchainStreamProviderTF[Future]

trait BlockchainStreamReaderTF[Future]

trait BlockchainProviderTF[Future]

trait BlockchainReaderTF[Future]

class Object

trait Matchable

class Any
Show all
Known subtypes: class StreamingEmulator

A streaming provider: BlockchainStreamReaderTF plus submission.

Attributes

Supertypes: trait BlockchainStreamReaderTF[F]

trait BlockchainProviderTF[F]

trait BlockchainReaderTF[F]

class Object

trait Matchable

class Any
Show all
Known subtypes: trait BlockchainStreamProvider

class StreamingEmulator

Future-based streaming reader, mirroring BlockchainReader.

Wanted for read-only surfaces — a scenario runner's post-run reader, an emulator projection — where handing out a provider would imply a submit path that does not exist.

Attributes

Supertypes: trait BlockchainReader

trait BlockchainStreamReaderTF[Future]

trait BlockchainReaderTF[Future]

class Object

trait Matchable

class Any
Show all
Known subtypes: trait BlockchainStreamProvider

class StreamingEmulator

Read-only streaming view of a blockchain: BlockchainReaderTF plus rollback-aware event subscriptions.

Stream / one-shot duality

Everything that changes over time has both a one-shot read and a subscription, and the two read the same state — a one-shot is semantically subscribeXxx().head. That is what stops two methods disagreeing about the current value. Polling disappears with it: pollForConfirmation(h) becomes subscribeTransactionStatus(h) with no sleep loop and no missed-update window.

Subscriptions are live when `subscribe` returns

Registration is eager and synchronous. Two consequences, both load-bearing:

subscribe(q) followed by submit(tx) on the same thread is race-free — the subscription is registered before subscribe returns, so the submitted transaction's events reach it. This is what makes emulator-driven tests deterministic.
The caller owns releasing the subscription. A subscription that is never consumed still accumulates events. Adapters expose this as a resource (fs2 Resource, pekko KillSwitch, an ox scope); the underlying release is ScalusAsyncSource.cancel().

Refusal

A request the provider cannot serve throws scalus.cardano.infra.UnsupportedSubscriptionException synchronously, from the call that caused it, before anything is registered — exactly when SubscriptionSupport.of(request, streamCapabilities) says Unsupported, or says Unindexed and the request did not set allowUnindexedScan. Callers that want to decide before asking consult the same function.

Choosing a stream type

The stream type is chosen per call, not per provider, so one provider instance can hand fs2 streams to application code while a test pulls raw ScalusAsyncSources from it. C is inferred from the expected type:

val tips: Stream[IO, ChainTip]        = provider.subscribeTip()  // needs the fs2 adapter given
val raw:  ScalusAsyncSource[ChainTip] = provider.subscribeTip()  // no adapter module needed

Where there is no expected type, C is decided by which adapter is in implicit scope — an imported adapter outranks the ScalusAsyncStreamAdapter.identity instance in this companion. Importing an adapter is therefore what makes subscribe return that library's type. Two adapters imported into one file with no expected type is an ambiguity, and the compiler says so; annotate or pass the type argument explicitly.

Type parameters

F: effect type for one-shot operations

Attributes

Supertypes: trait BlockchainReaderTF[F]

class Object

trait Matchable

class Any
Known subtypes: trait BlockchainStreamProviderTF[F]

trait BlockchainStreamProvider

class StreamingEmulator

trait BlockchainStreamReader

A position on the chain, identified by slot and block header hash.

Every event carries one, so subscribers can correlate across streams, deduplicate on replay, and checkpoint their own progress — a persisted ChainPoint is what StartFrom.At resumes from.

Attributes

Companion: object
Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: class
Supertypes: trait Product

trait Mirror

class Object

trait Matchable

class Any
Self type: ChainPoint.type

Attributes

Companion: class
Supertypes: trait Product

trait Mirror

class Object

trait Matchable

class Any
Self type: ChainTip.type

Chain tip: position plus height.

Height is what enables local confirmation arithmetic, so a subscriber tracking its own transaction can compute depth without another round trip.

Attributes

Companion: object
Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Overflow behaviour for a delta buffer.

There is no backpressure to be had here, and that is a property of the source rather than of this implementation: a chain produces blocks whether or not anyone is keeping up, so the only decisions available are how much to hold and what to do when that is not enough.

Chain-sourced events must never be dropped silently — a missed Spent corrupts the subscriber's view of state permanently, and it has no way to notice. So the two options are "never drop" and "fail loudly":

Bounded(n) — on overflow the subscription terminates with scalus.cardano.infra.ScalusBufferOverflowException, so the subscriber knows its view is untrustworthy and must resync. Default, see DeltaBufferPolicy.default.
Unbounded — memory is the only bound. For a subscriber that would rather be killed by the OOM killer than resync.

Deliberately not offered: a bounded buffer that drops. It would turn the corruption this policy exists to prevent into the default behaviour, and the subscriber could not detect it.

Attributes

Companion: object
Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: enum
Supertypes: trait Sum

trait Mirror

class Object

trait Matchable

class Any
Self type: DeltaBufferPolicy.type

What a provider can index a subscription by.

Mirrors the cases of scalus.cardano.node.UtxoSource: a query whose source reduces to kinds the provider declares here can be served from an index or a server-side filter; anything else needs a scan. UtxoSource.FromPaymentCredential deliberately has no matching case yet, so it is always treated as unindexed — see SubscriptionSupport.indexedSource.

Attributes

Companion: object
Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: enum
Supertypes: trait Sum

trait Mirror

class Object

trait Matchable

class Any
Self type: PushdownKind.type

How far back a provider can start a subscription.

Replay is not a yes/no property, for the same reason pushdown is not: on a REST backend an address's history is one paginated endpoint away while the chain's history is unreachable.

Attributes

Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

The consumer side of a subscription: pull the next event, or stop pulling.

Future is the lingua franca here on purpose. It is in the standard library, it exists on both JVM and JS, and it lets every adapter bridge its own effect — cats-effect IO, direct-style ox, pekko materialization — without any of those concepts leaking into this module.

Buffering lives behind this interface rather than in front of it: coalescing latest-value streams and failing bounded delta buffers on overflow are subtle enough that reimplementing them once per stream library would be several chances to get it wrong.

Attributes

Supertypes: class Object

trait Matchable

class Any
Known subtypes: trait Mailbox[A]

Typeclass for building a stream type S[_] from a ScalusAsyncSource.

One instance per stream library, each a handful of lines. The identity instance below means a caller who does not want to choose a library does not have to.

Attributes

Companion: object
Supertypes: class Object

trait Matchable

class Any
Known subtypes: object identity

Attributes

Companion: trait
Supertypes: class Object

trait Matchable

class Any
Self type: ScalusAsyncStreamAdapter.type

What happens to a subscription the provider cannot serve from an index.

The distinction between the first two is not how clever the matching is — it is whether the provider already holds the block's contents. An in-memory ledger, a chain-sync follower and a gRPC stream all see every transaction anyway, so a subscription that matches everything costs them nothing beyond the fan-out they already do. A REST provider whose cheap path is per-address endpoints has to fetch the block and then a UTxO set per transaction in it, which is the difference between a handful of requests a day and a spent quota.

The third is a different kind of answer, and it needs to be stated rather than approximated by the second: a provider built entirely out of per-source lookups has no request sequence that would answer "every transaction" at all. Calling that Metered would let a caller consent, via allowUnindexedScan, to something that cannot happen — and the subscription would then be accepted and deliver nothing, forever, with no error. Consenting to an impossibility is worse than being refused.

Attributes

Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Where a subscription starts reading.

Support is provider- and query-dependent — see StreamCapabilities and SubscriptionSupport. A provider that cannot serve the requested start point throws scalus.cardano.infra.UnsupportedSubscriptionException from subscribe.

Attributes

Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Everything a provider declares about its streaming behaviour.

This is the only thing an implementation states. Whether a particular subscription is supported, and whether it is cheap, is derived from this descriptor by SubscriptionSupport.of — so a provider cannot refuse something it advertised, or accept something it did not, and callers can decide what to ask for before asking.

Value parameters

idleSignals: whether the provider can emit progress signals for non-matching blocks
kinds: which subscription kinds are served at all
maxConfirmations: the largest confirmation depth the provider can gate on, if bounded
pushdown: query sources that can be served from an index rather than a scan
replay: how far back a subscription can start
rollbackHorizon: how deep a reorg the provider can report. None means it never signals rollbacks — which is not the same as "no rollbacks happen", and is exactly why it is declared rather than assumed
scanning: what a query outside pushdown costs — Free for a provider that already holds every block, Metered for one that would have to go and fetch it

Attributes

Companion: object
Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: class
Supertypes: trait Product

trait Mirror

class Object

trait Matchable

class Any
Self type: StreamCapabilities.type

Attributes

Companion: class
Supertypes: class Object

trait Matchable

class Any
Self type: StreamingEmulator.type

A streaming view of an scalus.cardano.node.EmulatorBase.

Every read delegates to the emulator, so the streaming and one-shot views cannot disagree — there is literally one ledger behind both. Submission additionally drives the subscription hub, so code that subscribes and reacts runs unchanged against a simulated chain.

One transaction, one block. The emulator has no notion of a block, so this wrapper synthesises one per accepted transaction and one per newEmptyBlock. Block identity is synthetic too — derived from the block number — which is why block subscriptions are declared unsupported rather than served with a fabricated header. UTxO and transaction subscriptions, which is what an application actually reacts to, are fully served.

Events only follow acceptance. A block reaches the hub only on Right(hash) from submitSync, that is, only after the emulator's validators and mutators have passed and the new state is committed. A rejected transaction produces no events.

Value parameters

securityParam: settlement depth this emulator claims. 0 — the default — matches its behaviour: a linear emulator never forks, so nothing ever needs to settle. Raise it to exercise a subscriber's confirmation gating.

Attributes

Companion: object
Supertypes: trait BlockchainStreamProvider

trait BlockchainProvider

trait BlockchainStreamReader

trait BlockchainReader

trait BlockchainStreamProviderTF[Future]

trait BlockchainStreamReaderTF[Future]

trait BlockchainProviderTF[Future]

trait BlockchainReaderTF[Future]

class Object

trait Matchable

class Any
Show all

Attributes

Companion: enum
Supertypes: trait Sum

trait Mirror

class Object

trait Matchable

class Any
Self type: SubscriptionKind.type

Which subscription kinds a provider serves at all.

Attributes

Companion: object
Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Per-subscription configuration for delta streams (UTxO, transaction, block).

Latest-value streams — tip, protocol params, transaction status — take no options: the provider always uses a size-1 coalescing buffer for them, because "the newest value wins" is the only sensible semantics for a single-source-of-truth cell.

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

A subscription a caller is considering, or is about to open.

Attributes

Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

The verdict on a subscription request.

Attributes

Companion: object
Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: enum
Supertypes: trait Sum

trait Mirror

class Object

trait Matchable

class Any
Self type: SubscriptionSupport.type

Transaction stream event.

Attributes

Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: enum
Supertypes: trait Sum

trait Mirror

class Object

trait Matchable

class Any
Self type: TransactionQuery.type

Which transactions a subscriber wants to observe.

The shape deliberately mirrors UTxORPC's TxPredicate (match / not / all_of / any_of), so a UTxORPC-backed provider lowers a query onto the wire predicate instead of translating between two different algebras — and so anything expressible here can be pushed down by the backends that support server-side filtering.

Attributes

Companion: object
Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

UTxO lifecycle event.

RolledBack is not an error and not an exception: a chain that reorganises is a chain behaving normally, and a subscriber that cannot express "undo everything after this point" has a bug waiting for its first fork. Providers that never signal rollbacks declare rollbackHorizon = None so the difference is visible before subscribing rather than after.

Attributes

Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Query for a UTxO event subscription.

Wraps a scalus.cardano.node.UtxoQuery — reusing its source/filter algebra and the inline DSL — and adds an event-type set, so a subscriber can ask for only Created events without a downstream collect.

The pagination fields on the inner query (limit, offset, minRequiredTotalAmount) are snapshot concepts and are ignored for subscriptions; they stay on UtxoQuery because BlockchainReader.findUtxos does honour them.

Attributes

Supertypes: trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

Attributes

Companion: enum
Supertypes: trait Sum

trait Mirror

class Object

trait Matchable

class Any
Self type: UtxoEventType.type

Which UTxO event kinds a subscription wants.

RolledBack and Idle are stream-wide signals rather than per-UTxO events, and are not selectable here — rollback delivery is governed by SubscriptionOptions.noRollback, idle delivery by SubscriptionOptions.idleSignals.

Attributes

Companion: object
Supertypes: trait Enum

trait Serializable

trait Product

trait Equals

class Object

trait Matchable

class Any
Show all

scalus.cardano.node.stream

Members list

Packages

Type members

Classlikes

Attributes

Attributes

Attributes

Attributes

Attributes

Stream / one-shot duality

Subscriptions are live when subscribe returns

Refusal

Choosing a stream type

Type parameters

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Value parameters

Attributes

Attributes

Attributes

Value parameters

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Attributes

Types

Attributes

Subscriptions are live when `subscribe` returns