Concrete network component that broadcasts messages to an etcd cluster and listens for incoming messages.

Get the client port corresponding to a listen address.

The client port used by the started etcd port is offset by the same amount as the listen address is offset by the default port 5001. This will result in the default client port 2379 be used by default still.

Derive the etcd client port from a configured peer (listen) port.

Exposed separately from getClientPort so test fixtures can pre-allocate both the peer and the derived client port without constructing a full NetworkConfiguration. Keep this and getClientPort in lockstep — any change to the offset must happen here, in one place.

Check and write version on etcd cluster. This will retry until we are on a majority cluster and succeed. If the version does not match a corresponding Connectivity message is sent via NetworkCallback.

Broadcast messages from a queue to the etcd cluster.

Retries on failure to putMessage in case we are on a minority cluster or when the grpc call timeouts.

Idempotent under transient GrpcDeadlineExceeded: putMessage uses an etcd transaction conditioned on the key's current mod_revision matching the last revision we successfully wrote. If a deadline-exceeded retry arrives at etcd after the original request already committed, the compare fails server-side (the mod_revision has advanced), the failure branch's range tells us what the new revision is, and we move on without writing a second time. So a retry whose original committed creates zero extra etcd revisions and the watcher on each peer sees exactly one event per logical broadcast. Same key namespace as master ('msg-<host>'), no disk growth.

Query etcd for the current mod_revision of this peer's broadcast key. Returns 0 if the key does not yet exist. Used by broadcastMessages to seed its in-memory baseline so subsequent compare mod_revision checks match etcd's reality from the very first putMessage.

Broadcast a message to the etcd cluster.

Wraps the etcd put in a Txn:

• compare: mod_revision(key) == lastModRev
• success: put(key, value)
• failure: range(key) (so we can learn the actual mod_revision if our compare failed, e.g. because a previous attempt of ours committed server-side after returning GrpcDeadlineExceeded to the client)

lastModRevVar is updated in both branches: from the response header on a successful put, from the range result on a compare failure. Retries of the same msg after a deadline-exceeded converge to a single effective revision rather than producing duplicate deliveries. Because broadcastMessages seeds lastModRev from etcd at startup (queryInitialModRev), a compare failure unambiguously means "this peer's earlier attempt already wrote a later revision than we have recorded" — i.e. the message is already delivered and the caller can pop.

Cluster-reset behaviour (intentionally fatal): if the compare fails and the range branch returns no kvs, the etcd cluster has lost the key we wrote against — either the data dir was wiped or the cluster was replaced underneath us. We do not silently reseed: this is a node-level event that should surface, not be papered over. putMessage calls fail, which kills the broadcast loop, propagates up to take down the node, and on restart queryInitialModRev re-seeds lastModRev from whatever state etcd actually has.

Assemble the etcd value for a batch of already-serialized messages: a CBOR list reusing each message's encoding verbatim. Uses the same indefinite-length list format as cardano-binary's list instances, so the value decodes as [msg] on the receiving side.

Fetch and wait for messages from the etcd cluster.

The persisted watch revision could not be read.

Fatal on purpose, and deliberately not a retryableEtcdError: re-reading the same bad file cannot succeed, and carrying on from revision 0 would rewind the watch to the beginning of history. Deleting the file is the fix, so say so.

Record the revision, atomically.

NOTE: Written to a temporary file and renamed into place, because a plain encodeFile is not atomic: killing the node mid-write (which happens routinely, both in tests and on restart) would otherwise leave a truncated file that getLastKnownRevision cannot parse.

NOTE: The rename is not fsynced, so an OS crash can still land it ahead of the data blocks. Closing that window costs an fsync per call, and this runs once per watch response, i.e. per received broadcast. The resulting file is instead handled on the next start, by LastKnownRevisionException and its hint.

Write a well-known key to indicate being alive, keep it alive using a lease and poll other peers entries to yield connectivity events. While doing so, overall network connectivity is determined from the ability to read/write to the cluster.

Predicate for gRPC errors that we treat as transient — connection blips or etcd-side disruption from which a retry is expected to recover. Anything outside this set is escalated by re-throwing.

GrpcNotFound is included specifically for lease loss: when etcd's RAFT leader changes under network stress, in-flight leases are revoked, and the next operation referencing one ('pollConnectivity.writeAlive') comes back with NOT_FOUND. The recovery is to mark the network as disconnected and let the outer loop recreate the lease, not crash the node.

Classify a failure talking to our local etcd, giving a reason to trace when retrying is the right response.

Connection-level failures qualify: etcd is a subprocess we started and connParams reconnects, so a lost connection is a blip rather than a reason to take the node down. Those arrive either as HTTP2Error (straight from the http2 client, e.g. the SETTINGS rate limit of #2817) or ServerDisconnected (grapesy's wrapper when a call outlives its connection). Everything else escalates, including putMessages cluster-reset fail and LastKnownRevisionException.

Add context to the grpcErrorMessage of any GrpcException raised.

Like withProcessTerm, but sends first SIGINT and only SIGTERM if not stopped within 5 seconds.

Queue elements carry the item's CBOR serialization, produced once on write and reused for both the on-disk file and the etcd value, so broadcasting does not serialize twice.

Create a new persistent queue at file path and given capacity.

Write a value to the queue, blocking if the queue is full.

Get the next value from the queue without removing it, blocking if the queue is empty.

Like peekPersistentQueue, but returns Nothing instead of blocking when the queue is empty.

Get all pending values and their serializations, up to the given count and total byte limits, blocking until at least one is available. Values are not removed; use popBatchPersistentQueue after they were sent.

Get the batch to broadcast next: the batch already in flight if there is one, otherwise a fresh one peeked from the queue (and recorded as in flight). Returns Nothing when nothing is pending. The caller must clear the in-flight var after popping a successfully sent batch.

Pinning the in-flight batch across transient retries matters for the compare-fail dedup in putMessage: a put can commit server-side while the client sees e.g. GrpcDeadlineExceeded. The retry must send (and afterwards pop) exactly the content of the committed attempt. Re-peeking on retry could pick up messages enqueued in the meantime; the dedup branch would then declare the grown batch delivered and the never-sent tail would be popped and lost.

Remove a batch previously returned by peekBatchPersistentQueue. Pops unconditionally, one item per batch entry: this thread is the sole consumer, so the queue head still holds exactly the peeked items (an item-matching guard could only silently no-op and wedge the queue, see #2742).

Delete the backing file of a popped queue item. Failing to delete is traced but not fatal: the message was already broadcast, so a leftover file only means it may be re-broadcast after a restart (at-least-once delivery, same as the crash-recovery path).