Feature: zmq realtime events

[MorningLightMountain713]

Overview

This PR introduces real-time event notifications for FluxNode state changes via ZMQ pub/sub, enabling efficient event-driven architectures for monitoring and validation tools.

FluxOS gravity uses a lot of polling with fluxd - which is very heavy and uses a lot of resources for both fluxd and gravity.

New ZMQ Endpoints

Four new ZMQ publication endpoints provide real-time notifications:

• zmqpubhashblockheight - Published when a new block is connected, includes block hash and height (36 bytes)
• zmqpubchainreorg - Published when a chain reorganization occurs, includes old/new tips and fork point (108 bytes)
• zmqpubfluxnodelistdelta - Published when FluxNode list changes, includes incremental state changes with block context (73+ bytes)
• zmqpubfluxnodestatus - Published when the local fluxnode's status changes (confirmed, paid, expired, etc.). Only fires on change, not every block. Non-fluxnodes skip with a single bool check. (54+ bytes)

Configure in flux.conf:

zmqpubhashblockheight=tcp://127.0.0.1:16123
zmqpubchainreorg=tcp://127.0.0.1:16123
zmqpubfluxnodelistdelta=tcp://127.0.0.1:16123
zmqpubfluxnodestatus=tcp://127.0.0.1:16123

Note: zmqpubfluxnodestatus is only useful on fluxnodes (fluxnode=1). It eliminates the need for FluxOS/gravity to poll getfluxnodestatus RPC.

Event-Based Architecture Benefits

Traditional polling approaches require clients to repeatedly request full snapshots to detect changes, creating unnecessary load on both client and server. Event-based notifications provide several advantages:

• Efficiency: Clients receive only the data that changed, reducing bandwidth and processing overhead
• Latency: Changes are pushed immediately rather than waiting for the next poll interval
• Scalability: Pub/sub architecture allows multiple subscribers without additional server load per client
• Precision: Delta messages include exact block context, eliminating ambiguity about which block state applies to

FluxNode List Delta Messages

Delta messages provide incremental state updates with full block context to ensure consistency.

Message Structure

Each delta message includes a 73-byte header followed by the changes:

[from_height: 4 bytes][to_height: 4 bytes]
[from_hash: 32 bytes][to_hash: 32 bytes]
[flags: 1 byte (bit 0 = is_reorg)]
[added_nodes: CompactSize + node data]
[removed_nodes: CompactSize + outpoints]
[updated_nodes: CompactSize + node data]

Block Context

The header provides complete block context:

• from_height / from_hash: State before this delta
• to_height / to_hash: State after this delta
• flags: Bit 0 indicates whether this delta was triggered by a chain reorganization

This enables clients to:

• Verify deltas chain correctly (next delta's from_hash matches previous to_hash)
• Detect gaps in the delta stream
• Identify when a delta applies to a fork vs main chain
• Recover from reorgs by comparing block hashes
• Distinguish reorg deltas from normal deltas via the is_reorg flag

Relationship to Snapshots

The getfluxnodesnapshot RPC provides the complete state at a specific block, while deltas provide incremental changes between blocks. Clients can:

1. Fetch an initial snapshot: getfluxnodesnapshot returns state at height H with blockhash B
2. Apply deltas: Each subsequent delta advances state by one block

Block Hash Validation

Each delta includes full block hashes, allowing verification that:

• The delta applies to the expected chain state
• No gaps exist in the delta sequence
• Chain reorganizations are detected immediately

If validation fails (hash mismatch or gap detected), the client can re-sync from a fresh snapshot.

FluxNode Status Messages

Status messages notify when the local fluxnode's state changes. Six fields are cached and compared each block — a message is only published when something changes:

[block_height: 4][status: 1][tier: 1]
[confirmed_height: 4][last_confirmed_height: 4][last_paid_height: 4]
[txhash: 32 reversed][outidx: 4]
[ip: CompactSize + string bytes]

Status values: 0=ERROR, 1=STARTED, 2=DOS_PROTECTION, 3=CONFIRMED, 4=MISS_CONFIRMED, 5=EXPIRED.

Performance

• Non-fluxnodes: Single fFluxnode bool check per block — zero cost
• Fluxnodes: GetFluxnodeData does 3 hashmap count() calls + 1 at() per block. Compares 6 fields. Publishes only on change (rare — payment every ~few hundred blocks)

Demo Python client:

Demo Python Network State validation (snapshot vs deltas):

Integration Tests

Comprehensive integration tests verify correct behavior across all scenarios:

Test Coverage

• test_hashblockheight: Verifies block notifications include correct hash and height
• test_fluxnodelistdelta: Validates delta structure and that changes are reflected correctly
• test_delta_consistency: Confirms deltas chain correctly and match snapshot data
• test_chainreorg: Verifies reorg notifications and delta behavior across forks

Chain Reorganization Testing

The reorg test creates a network partition:

1. Split network into two groups
2. Each group mines competing chains using setmocktime to force divergence
3. Rejoin network to trigger reorganization
4. Verify reorg notification includes correct old/new tips and fork point
5. Confirm delta stream continues correctly after reorg

Test Execution

All tests pass consistently:

BITCOIND=../../src/fluxd BITCOINCLI=../../src/flux-cli uv run --with pyzmq fluxnode_zmq_test.py --nocleanup

The chainreorg test runs last to avoid interference with other tests, as it deliberately creates inconsistent network state.

Monitoring and Validation Tools

Two production-ready tools are included in contrib/zmq/:

flux-zmq-monitor

Real-time monitoring package that subscribes to all ZMQ events and displays them in human-readable format.

Features:

• Decodes binary message formats for all four event types
• Displays FluxNode changes with tier, IP, status, and payment info
• Handles delta header format with block hashes and reorg flag
• Includes systemd service for production deployment

flux-state-validator

Production validator that continuously verifies FluxNode state consistency.

What It Validates:

• Delta messages chain correctly (from_hash matches previous to_hash)
• Snapshots are atomic (height and blockhash from same block)
• State can be reconstructed from deltas
• No gaps or out-of-order messages
• Chain reorganizations are handled correctly

Race Prevention:

• Buffers deltas that arrive during initialization
• Validates block hashes to detect gaps or forks
• Automatically re-syncs if inconsistency detected

Deployment

Both tools include hardened systemd service files:

• Security features: NoNewPrivileges, ProtectSystem=strict, ProtectHome=true
• Soft dependency on fluxd (Wants=) so ZMQ auto-reconnects survive daemon restarts
• Automatic log directory creation
• UV package manager integration
• Comprehensive documentation in contrib/zmq/README.md

[MorningLightMountain713]

Just working through a bug in what I believe is the deterministic list RPC that is causing a mismatch with deltas

[MorningLightMountain713]

Just working through a bug in what I believe is the deterministic list RPC that is causing a mismatch with deltas

Yeah, it's an existing bug. Will fix

[MorningLightMountain713]

I believe this is now fixed... will monitor deltas for a few days.

Now we don't need to aquire / release the lock on every single GetFluxnodeData call (about 7.5k calls currently) and only does 1 lookup instead of 3. This follows the same pattern as GetNextPayment

This was quite a hot path as Gravity calls this every 30 seconds.

This stops nodes from being included in the viewdeterministicfluxnodelist RPC (and new snapshot RPC) that have expired AND have a new start tx. This was also causing all nodes after the expired node(s) to have the incorrect rank, as the expired node still held a rank which was incorrect - in reality it was getting discarded when it gets to the front of the queue, so every node after the "fantom" node had their rank inflated by one (for however many fantom nodes were in the list).

When the "fantom" node was confirmed, it gets removed and readded to the end of the list (which makes sense).

[MorningLightMountain713]

Looks good, delta updates seem consistent now and match the snapshots. Has handled a couple of reorgs too

[MorningLightMountain713]

Has been running 4 days. Looks good:

Some gdb analsys of memory usage etc:

  ZMQ Extension Memory & Efficiency Report

  System: charlie (fluxd PID 3864144, uptime 3.8 days)
  Binary BuildID: 6e201ae610608fd7ab5f8924db60bdc95fea7e09 (confirmed match)
  Block height: 2,360,261 | Network nodes: 7,863
  Validator stats: 11,545 deltas applied, 1,153 validations, 100% success rate, 6
  reorgs handled

  ---
  1. Delta Maps (g_fluxnodeDelta) — NO LEAK

  ┌────────────┬──────────────┬──────────────────┐
  │    Map     │ Count (live) │     Expected     │
  ├────────────┼──────────────┼──────────────────┤
  │ mapAdded   │ 0            │ 0 between blocks │
  ├────────────┼──────────────┼──────────────────┤
  │ setRemoved │ 0            │ 0 between blocks │
  ├────────────┼──────────────┼──────────────────┤
  │ mapUpdated │ 0            │ 0 between blocks │
  └────────────┴──────────────┴──────────────────┘

  The delta maps are correctly cleared after each block via g_fluxnodeDelta.Clear()
   inside SendDelta(). After 11,545+ blocks of operation, they hold zero entries.
  No memory leak.

  2. Struct Sizes (from GDB)

  ┌────────────────────────────────────┬──────────────┐
  │             Structure              │ Size (bytes) │
  ├────────────────────────────────────┼──────────────┤
  │ FluxNodeDelta (global)             │ 184          │
  ├────────────────────────────────────┼──────────────┤
  │ FluxnodeCacheData (per node)       │ 280          │
  ├────────────────────────────────────┼──────────────┤
  │ COutPoint (per outpoint)           │ 36           │
  ├────────────────────────────────────┼──────────────┤
  │ CZMQPublishFluxNodeListNotifier    │ 104          │
  ├────────────────────────────────────┼──────────────┤
  │ CZMQPublishFluxNodeStatusNotifier  │ 144          │
  ├────────────────────────────────────┼──────────────┤
  │ CZMQAbstractPublishNotifier (base) │ 88           │
  ├────────────────────────────────────┼──────────────┤
  │ FluxnodeCache                      │ 968          │
  └────────────────────────────────────┴──────────────┘

  3. Permanent Memory Overhead of Our Changes

  Fixed (one-time) allocations:

  ┌─────────────────────────────────────────────────────────────┬────────┐
  │                            Item                             │ Bytes  │
  ├─────────────────────────────────────────────────────────────┼────────┤
  │ g_fluxnodeDelta (global struct, 3 empty containers + mutex) │ 184    │
  ├─────────────────────────────────────────────────────────────┼────────┤
  │ CZMQPublishFluxNodeListNotifier instance                    │ 104    │
  ├─────────────────────────────────────────────────────────────┼────────┤
  │ CZMQPublishFluxNodeStatusNotifier instance                  │ 144    │
  ├─────────────────────────────────────────────────────────────┼────────┤
  │ CZMQPublishHashBlockHeightNotifier instance                 │ ~88    │
  ├─────────────────────────────────────────────────────────────┼────────┤
  │ CZMQPublishChainReorgNotifier instance                      │ ~88    │
  ├─────────────────────────────────────────────────────────────┼────────┤
  │ mapPublishNotifiers entries (4 string+ptr pairs)            │ ~320   │
  ├─────────────────────────────────────────────────────────────┼────────┤
  │ ZMQ socket (1 PUB socket shared via tcp://127.0.0.1:16123)  │ ~16 KB │
  ├─────────────────────────────────────────────────────────────┼────────┤
  │ Total fixed overhead                                        │ ~17 KB │
  └─────────────────────────────────────────────────────────────┴────────┘

  Transient (per-block) allocations:

  Typical delta per block from the validator log: +0 -0 ~15 (about 15 updated nodes
   per block, occasionally 1-3 added/removed).

  Phase: Delta accumulation (Flush → SendDelta)
  Peak memory: ~15 × (36 + 280) = ~4.7 KB
  Duration: milliseconds
  ────────────────────────────────────────
  Phase: CDataStream serialization buffer
  Peak memory: ~73 bytes header + 15 × ~200 bytes = ~3.1 KB
  Duration: milliseconds
  ────────────────────────────────────────
  Phase: Peak transient
  Peak memory: ~8 KB
  Duration: freed every block

  These are freed immediately after SendDelta() calls Clear() and the CDataStream
  goes out of scope. Confirmed at zero between blocks by GDB.

  4. Process Memory Context

  ┌─────────────────────────┬──────────┐
  │         Metric          │  Value   │
  ├─────────────────────────┼──────────┤
  │ VmRSS (resident)        │ 1,862 MB │
  ├─────────────────────────┼──────────┤
  │ VmPeak                  │ 2,411 MB │
  ├─────────────────────────┼──────────┤
  │ VmHWM (high water mark) │ 1,867 MB │
  ├─────────────────────────┼──────────┤
  │ VmSwap                  │ 0        │
  ├─────────────────────────┼──────────┤
  │ Anon (heap+stack)       │ 1,798 MB │
  ├─────────────────────────┼──────────┤
  │ File-backed (mmap)      │ 66 MB    │
  ├─────────────────────────┼──────────┤
  │ Threads                 │ 34       │
  ├─────────────────────────┼──────────┤
  │ FDs (sockets)           │ 65       │
  └─────────────────────────┴──────────┘

  Our changes add ~17 KB fixed + ~8 KB transient to a 1.8 GB process. That's 0.001%
   — completely negligible.

[MorningLightMountain713]

I have been running this on my BBB (Big Beautiful Branch) which includes my 5 PRs:

#276
#275
#274
#273
#271

All seems to work well.