pub enum ConnectionEvent {
Show 31 variants
Connected {
protocol_version: i32,
max_message_size: u32,
feature_flags: FeatureFlags,
},
AuthChallenge {
method: Option<String>,
challenge: Option<Bytes>,
},
ProducerReady {
handle: ProducerHandle,
producer_name: String,
last_sequence_id: i64,
schema_version: Bytes,
},
ProducerOpenFailed {
handle: ProducerHandle,
code: i32,
message: String,
},
ProducerOpenFailedTransient {
handle: ProducerHandle,
code: i32,
message: String,
},
SubscribeAcked {
handle: ConsumerHandle,
},
SubscribeFailed {
handle: ConsumerHandle,
code: i32,
message: String,
},
SubscribeFailedTransient {
handle: ConsumerHandle,
code: i32,
message: String,
},
Message {
handle: ConsumerHandle,
message: IncomingMessage,
},
MessageReceivedFromShadow {
handle: ConsumerHandle,
source_topic: String,
source_message_id: MessageId,
shadow_message_id: MessageId,
message: IncomingMessage,
},
ReplicatedSubscriptionMarkerObserved {
handle: ConsumerHandle,
marker: ReplicatedSubscriptionMarker,
},
SendReceipt {
handle: ProducerHandle,
sequence_id: SequenceId,
message_id: MessageId,
},
SendError {
handle: ProducerHandle,
sequence_id: SequenceId,
code: i32,
message: String,
},
AckResponse {
request_id: Option<RequestId>,
result: Result<(), String>,
},
LookupResponse {
request_id: RequestId,
result: LookupOutcome,
},
PartitionedMetadataResponse {
request_id: RequestId,
partitions: u32,
error: Option<(i32, String)>,
},
TopicListSnapshot {
request_id: RequestId,
topics: Vec<String>,
},
TopicListChanged {
added: Vec<String>,
removed: Vec<String>,
},
ReachedEndOfTopic {
handle: ConsumerHandle,
},
ActiveConsumerChanged {
handle: ConsumerHandle,
active: bool,
},
ConsumerStalled {
handle: ConsumerHandle,
permit_balance: u32,
stalled_for: Duration,
},
TopicMigrated {
producer: Option<ProducerHandle>,
consumer: Option<ConsumerHandle>,
broker_service_url: Option<String>,
broker_service_url_tls: Option<String>,
},
RedirectUrlRejected {
source: &'static str,
broker_service_url: Option<String>,
broker_service_url_tls: Option<String>,
},
ProducerClosedByBroker {
handle: ProducerHandle,
assigned_broker_service_url: Option<String>,
},
ConsumerClosedByBroker {
handle: ConsumerHandle,
assigned_broker_service_url: Option<String>,
},
ChecksumMismatch {
computed: u32,
expected: u32,
},
Closed {
reason: Option<String>,
},
TxnResponse {
request_id: RequestId,
outcome: TxnRoundTrip,
},
GetSchemaResponse {
request_id: RequestId,
result: Result<(Schema, Option<Bytes>), (i32, String)>,
},
AntiThrashCooldown {
until: Instant,
},
AntiThrashCleared,
}Expand description
A semantic event surfaced by the state machine.
Variants§
Connected
Handshake completed. The driver should now allow producer/consumer opens.
Fields
feature_flags: FeatureFlagsCapabilities negotiated with the broker.
AuthChallenge
The broker sent a CommandAuthChallenge mid-connection.
The auth layer (above magnetar-proto) is expected to compute the response and feed
it back via Connection::submit_auth_response.
Fields
ProducerReady
A producer that was queued via create_producer is now ready to send.
Fields
handle: ProducerHandleThe producer handle this event refers to.
ProducerOpenFailed
The broker rejected a CommandProducer open with CommandError.
Emitted from the CommandError handler when the failing request id correlates with a
pending producer-open. The corresponding producer state has already been dropped from
the connection — the user-facing handle is dead. Pair with
ConnectionEvent::ProducerReady as the success/failure split for an open_producer
round-trip.
Fields
handle: ProducerHandleThe producer that failed to open.
ProducerOpenFailedTransient
An established producer’s reattachment was rejected with a retryable ADR-0080 code.
This event is emitted only after the producer has attached successfully
at least once. The state is retained and the runtime driver must back
off, re-run lookup, and call
crate::Connection::retry_producer_open. A provisional first-open
rejection instead emits Self::ProducerOpenFailed and is retried by
the routing-aware client.
Fields
handle: ProducerHandleThe producer that failed to open.
SubscribeAcked
A subscribe request was acknowledged by the broker.
Fields
handle: ConsumerHandleThe consumer handle this event refers to.
SubscribeFailed
The broker rejected a CommandSubscribe with CommandError.
Emitted from the CommandError handler when the failing request id correlates with a
pending subscribe. The corresponding consumer state has already been dropped from the
connection. Pair with ConnectionEvent::SubscribeAcked as the success/failure split
for a subscribe round-trip.
Fields
handle: ConsumerHandleThe consumer that failed to subscribe.
SubscribeFailedTransient
Established-consumer companion to
Self::ProducerOpenFailedTransient.
The consumer has attached successfully at least once, so its state is
retained for driver-owned reattachment through
crate::Connection::retry_consumer_subscribe. A provisional
subscribe rejection emits Self::SubscribeFailed and is retried by
the routing-aware client.
Fields
handle: ConsumerHandleThe consumer that failed to subscribe.
Message
An incoming message was delivered by the broker.
Fields
handle: ConsumerHandleThe consumer that received it.
message: IncomingMessageThe decoded message.
MessageReceivedFromShadow
PIP-180 / ADR-0033: an incoming message was delivered by the broker on a
shadow topic, originating from a source topic. Emitted in place of
Self::Message when the consumer was subscribed to a shadow topic
(resolved at subscribe time via the admin REST getShadowTopics(source)
hint, see crate::consumer::ConsumerState::set_shadow_metadata) AND
the inbound entry’s pb::MessageMetadata::replicated_from is set.
source_message_id and message.message_id compare equal under the
PIP-180 structural-equality contract documented on
crate::types::MessageId — the broker presents shadow-side entries
with the source-topic (ledger_id, entry_id, batch_index, partition),
so cross-side deduplication needs no out-of-band correlation key.
Callers that don’t care about the shadow context can collapse this
variant onto Self::Message by inspecting message. The variant
is non-breaking by convention (ConnectionEvent is treated as
#[non_exhaustive] per ADR-0033’s “new sum-variant is additive” risk
note).
Fields
handle: ConsumerHandleThe consumer that received it.
source_topic: StringSource-topic name (resolved via admin REST getShadowTopics(source)
at subscribe time, cached on
crate::consumer::ConsumerState::shadow_metadata).
source_message_id: MessageIdSource-topic MessageId. Equal to message.message_id under
crate::types::MessageId’s structural-equality contract.
shadow_message_id: MessageIdShadow-side MessageId — same fields as source_message_id, but
surfaced separately so callers don’t have to derive it from
message.
message: IncomingMessageThe decoded message — same payload + metadata the consumer would
have surfaced via Self::Message on a non-shadow topic.
ReplicatedSubscriptionMarkerObserved
PIP-33: the broker emitted a REPLICATED_SUBSCRIPTION_* marker on this
consumer’s topic. Surfaced for observability only — the marker is filtered
off the user-visible message stream (never appears as Self::Message)
because it carries broker-side snapshot/update payload, not application
data. Magnetar never originates these markers; the broker generates them
when the namespace has replicated_subscription_status=true and a peer
cluster’s snapshot/update cycle fires. See
crate::markers for the payload typing and
ADR-0034
for scope.
Fields
handle: ConsumerHandleThe consumer that received the marker.
marker: ReplicatedSubscriptionMarkerDecoded marker payload (kind + details).
SendReceipt
A CommandSendReceipt correlated with one of our pending publishes.
Fields
handle: ProducerHandleThe producer that owns the publish.
sequence_id: SequenceIdPublisher-side sequence id of the receipt.
SendError
A CommandSendError correlated with one of our pending publishes.
Fields
handle: ProducerHandleThe producer that owns the publish.
sequence_id: SequenceIdPublisher-side sequence id of the failed publish.
AckResponse
Response to a CommandAck request.
Fields
LookupResponse
Response to a CommandLookupTopic request.
Fields
result: LookupOutcomeResolved broker URL on success, None on failure or redirect.
PartitionedMetadataResponse
Response to a CommandPartitionedTopicMetadata request.
Fields
TopicListSnapshot
Topic list watcher initial snapshot.
Fields
TopicListChanged
Topic list watcher delta (PIP-145).
Fields
ReachedEndOfTopic
The broker signalled end-of-topic on a non-durable subscription.
Fields
handle: ConsumerHandleThe consumer that reached end-of-topic.
ActiveConsumerChanged
A consumer’s active/passive state changed (failover).
Fields
handle: ConsumerHandleThe consumer whose active state changed.
ConsumerStalled
A consumer has held un-spent broker permits over an empty receive queue, in a
dispatch-eligible state, for
ConnectionConfig::consumer_stall_timeout
without a single dispatch unit arriving (issue #414).
Emitted at most once per stall episode: the next dispatch unit re-arms the
watchdog, so a consumer that recovers and wedges again reports twice. None of
the client’s own state explains the silence — not pause, not an in-flight seek,
not end-of-topic, not a re-attach in progress; all of those suppress the watchdog.
The wire protocol carries only monotonic client → broker permit increments
(CommandFlow), so the client cannot itself drive the broker’s counter negative:
this event says the BROKER stopped dispatching against a grant it acknowledged.
The connection is otherwise healthy — ADR-0058’s keepalive keeps passing, which is
precisely why it cannot detect this.
Purely diagnostic. Recovery is the caller’s explicit choice:
Connection::resubscribe_consumer_in_place
repairs this client’s own dispatcher slot; a dispatcher-wide broker corruption
needs an operator-side pulsar-admin topics unload.
Fields
handle: ConsumerHandleThe consumer whose dispatch went silent.
TopicMigrated
Broker requested the producer or consumer to migrate to a different broker URL.
Fields
producer: Option<ProducerHandle>Producer handle if the resource type was Producer.
consumer: Option<ConsumerHandle>Consumer handle if the resource type was Consumer.
RedirectUrlRejected
The broker advertised a redirect URL (PIP-188 TopicMigrated,
CommandLookupTopicResponse proxy URL, CommandCloseProducer /
CommandCloseConsumer reassignment URL, …) that the configured
RedirectUrlAllowList rejected.
The runtime engines surface this event instead of honouring the
URL — no CommandConnect is sent to the rejected host under the
original AuthProvider. Defence in
depth: a compromised broker (or a MITM downstream of TLS
termination) cannot harvest the reused credentials by advertising
an attacker-controlled URL.
urls carries the rejected URL list (plain + TLS variants, in
that order, both Option<String>) so operators can audit the
trigger. The runtime keeps using its original URL.
Fields
ProducerClosedByBroker
The broker asked us to close a producer (e.g. fenced).
Fields
handle: ProducerHandleThe producer that was closed.
ConsumerClosedByBroker
The broker asked us to close a consumer.
Fields
handle: ConsumerHandleThe consumer that was closed.
ChecksumMismatch
A CRC32C checksum mismatch was detected on an inbound payload frame.
Per [GUIDELINES.md] §“Protocol-correctness invariants”, the frame is dropped (never delivered) and the event is surfaced for diagnostics.
Closed
The connection is closing (locally initiated or peer-triggered).
TxnResponse
A Transaction Coordinator round-trip completed.
Carries the outcome for one of: new_txn, add_partition_to_txn,
add_subscription_to_txn, end_txn.
Fields
outcome: TxnRoundTripThe transactional outcome.
GetSchemaResponse
Response to a CommandGetSchema request (PIP-87 broker-side schema lookup).
Emitted after the runtime calls
Connection::get_schema and the broker replies. The
payload is a GetSchemaResult — Ok carries the registry-resolved pb::Schema and
schema version, Err carries the broker’s ServerError code and message.
Fields
AntiThrashCooldown
The connection-level anti-thrash detector (ADR-0028) has engaged. The
supervisor must sleep until until before its next
Transport::connect, even if its per-handle backoff would have
retried sooner. Emitted exactly once on each Normal → Cooldown
transition by
Connection::record_reattach_outcome.
Fields
AntiThrashCleared
The connection-level anti-thrash cooldown (ADR-0028) has lifted —
either because the supervisor slept past until and explicitly
cleared it via
Connection::anti_thrash_state_mut,
or because the broker stabilised and the driver called
Connection::record_first_op_success.
Trait Implementations§
Source§impl Clone for ConnectionEvent
impl Clone for ConnectionEvent
Source§fn clone(&self) -> ConnectionEvent
fn clone(&self) -> ConnectionEvent
1.0.0 (const: unstable) · Source§fn clone_from(&mut self, source: &Self)
fn clone_from(&mut self, source: &Self)
source. Read more