node: publish Core's sequence ZMQ topic for block connects and disconnects

CONCEPTS.md

@@ -137,9 +137,21 @@ keeps its ownership for retry.
 
 Still open around it: returning a disconnected block's transactions through one
 production admission pipeline shared by Electrum, P2P relay, and reorg handling;
-publishing a disconnect notification; and backfilling the filter index after a
-gap. Raw mempool insertion is not reconsideration because it cannot reconstruct
-fee, policy, conflict, and ancestry metadata.
+and backfilling the filter index after a gap. The `pubsequence` stream publishes
+block connect/disconnect notifications, but intentionally does not publish
+mempool `A`/`R` events: the current mempool counter and mutation reasons cannot
+yet guarantee the enforcer's required contiguous transaction event sequence.
+Raw mempool insertion is not reconsideration because it cannot reconstruct fee,
+policy, conflict, and ancestry metadata.
+
+### Sequence stream
+
+The Core-compatible `pubsequence` ZMQ stream is a unified block-event stream.
+Each event carries the block hash, one label (`C` for connect or `D` for
+disconnect), and a topic-local little-endian `u32` sequence counter. Reorg
+disconnects are emitted tip-first before connects on the replacement branch.
+This implementation deliberately omits mempool `A`/`R` events until the
+mempool has per-transaction sequence assignment and explicit removal reasons.
 
 ### Dispatch-bound parallelism
 

README.md

@@ -150,8 +150,9 @@ header tip against the applied tip each tick and switches branches when the
 applied chain is outweighed.
 
 It is still not the node to depend on. A disconnected block's transactions do
-not return to the mempool, no disconnect notification is published, and the
-filter index is not backfilled across a gap. `docs/README.md` lists the rest.
+not return to the mempool, and the filter index is not backfilled across a gap.
+The ZMQ `pubsequence` stream now publishes block connect/disconnect events, but
+does not emit mempool `A`/`R` events. `docs/README.md` lists the rest.
 
 ## Documentation
 

crates/node/src/apply.rs

@@ -1047,7 +1047,9 @@ pub fn disconnect_block(
     let transition = handles
         .begin_chain_transition()
         .map_err(|error| crate::DisconnectError::Refused(Box::new(error)))?;
-    disconnect_block_admitted(handles, block, &transition)
+    let result = disconnect_block_admitted(handles, block, &transition);
+    drop(transition);
+    result
 }
 
 /// Disconnects one block while the caller holds admission and `chain_transition`.
@@ -1217,6 +1219,14 @@ pub(crate) fn disconnect_block_admitted(
             })
         })?;
 
+    if handles.zmq_publisher.wants_notifications() {
+        handles
+            .zmq_publisher
+            .publish_sequence(crate::zmq_publisher::SequenceEvent::Disconnected(
+                block_hash,
+            ));
+    }
+
     // The marker deliberately stays set here.
     //
     // Every mutation landed, but only some of them are durable. The index
@@ -2315,6 +2325,9 @@ fn apply_block_admitted(
         // Best-effort ZMQ event emission. Failures must not propagate per the
         // ZmqPublisher contract; the trait's methods return `()`.
         handles.zmq_publisher.publish_hashblock(tip.hash);
+        handles
+            .zmq_publisher
+            .publish_sequence(crate::zmq_publisher::SequenceEvent::Connected(tip.hash));
         if wants_rawblock {
             handles.zmq_publisher.publish_rawblock(&block_bytes);
         }
@@ -8678,6 +8691,128 @@ mod consensus_rule_tests {
         Ok(())
     }
 
+    #[derive(Debug, Default)]
+    struct RecordingSequencePublisher {
+        events: Mutex<Vec<(Hash256, u8, u32)>>,
+        next_sequence: Mutex<u32>,
+    }
+
+    impl crate::ZmqPublisher for RecordingSequencePublisher {
+        fn publish_hashblock(&self, _hash: Hash256) {}
+
+        fn publish_hashtx(&self, _txid: bitcoin::Txid) {}
+
+        fn publish_rawblock(&self, _bytes: &[u8]) {}
+
+        fn publish_rawtx(&self, _bytes: &[u8]) {}
+
+        fn publish_sequence(&self, event: crate::SequenceEvent) {
+            let (hash, label) = match event {
+                crate::SequenceEvent::Connected(hash) => (hash, b'C'),
+                crate::SequenceEvent::Disconnected(hash) => (hash, b'D'),
+            };
+            let mut next_sequence = self.next_sequence.lock();
+            self.events.lock().push((hash, label, *next_sequence));
+            *next_sequence += 1;
+        }
+    }
+
+    #[test]
+    fn reorg_sequence_events_disconnect_old_tip_before_connecting_new_branch()
+    -> Result<(), Box<dyn std::error::Error>> {
+        let utxo = Arc::new(UtxoSet::new());
+        let publisher = Arc::new(RecordingSequencePublisher::default());
+        let publisher_handle: Arc<dyn crate::ZmqPublisher> = publisher.clone();
+        let mut handles = apply_handles_without_tx_index(Network::Regtest, Arc::clone(&utxo))
+            .with_zmq_publisher(publisher_handle);
+        let bodies = Arc::new(MapBodyStore::default());
+        let body_handle: Arc<dyn crate::apply::PruneBodyStore> = bodies.clone();
+        handles.block_body_store = Some(body_handle);
+
+        let genesis = bitcoin::blockdata::constants::genesis_block(bitcoin::Network::Regtest);
+        let genesis_hash = Hash256::from_le_bytes(genesis.block_hash().as_byte_array());
+        let genesis_tip = applied_header_tip(&handles, genesis_hash, &genesis, 0)?;
+        handles.applied_tip.store(Some(Arc::new(genesis_tip)));
+
+        let old_one = mined_block_with_prev_hash_and_transactions(
+            genesis.block_hash(),
+            vec![coinbase_transaction(1)],
+        )?;
+        let old_one_raw = bytes::Bytes::from(bitcoin::consensus::encode::serialize(&old_one));
+        let old_one_tip = apply_block_with_serialized(&handles, &old_one, old_one_raw.clone())?;
+        bodies
+            .bodies
+            .write()
+            .insert((old_one_tip.height, old_one_tip.hash), old_one_raw.to_vec());
+
+        let old_two = mined_block_with_prev_hash_and_transactions(
+            old_one.block_hash(),
+            vec![coinbase_transaction(2)],
+        )?;
+        let old_two_raw = bytes::Bytes::from(bitcoin::consensus::encode::serialize(&old_two));
+        let old_two_tip = apply_block_with_serialized(&handles, &old_two, old_two_raw.clone())?;
+        bodies
+            .bodies
+            .write()
+            .insert((old_two_tip.height, old_two_tip.hash), old_two_raw.to_vec());
+        publisher.events.lock().clear();
+        *publisher.next_sequence.lock() = 0;
+
+        let new_one = mined_block_with_prev_hash_and_transactions(
+            genesis.block_hash(),
+            vec![coinbase_transaction(3)],
+        )?;
+        let new_two = mined_block_with_prev_hash_and_transactions(
+            new_one.block_hash(),
+            vec![coinbase_transaction(4)],
+        )?;
+        let target = {
+            let mut tree = handles.block_tree.write();
+            let mut target = None;
+            for (height, block) in [(1_u32, &new_one), (2_u32, &new_two)] {
+                target = Some(tree.insert_header(block.header, NodeStatus::HeaderValid)?);
+                bodies.bodies.write().insert(
+                    (
+                        height,
+                        Hash256::from_le_bytes(block.block_hash().as_byte_array()),
+                    ),
+                    bitcoin::consensus::encode::serialize(block),
+                );
+            }
+            target.ok_or_else(|| anyhow::anyhow!("new branch has no target"))?
+        };
+
+        crate::reorg::switch_to_branch(&handles, target, |_| None, |_| {})?;
+
+        let events = publisher.events.lock().clone();
+        assert_eq!(
+            events,
+            vec![
+                (
+                    Hash256::from_le_bytes(old_two.block_hash().as_byte_array()),
+                    b'D',
+                    0
+                ),
+                (
+                    Hash256::from_le_bytes(old_one.block_hash().as_byte_array()),
+                    b'D',
+                    1
+                ),
+                (
+                    Hash256::from_le_bytes(new_one.block_hash().as_byte_array()),
+                    b'C',
+                    2
+                ),
+                (
+                    Hash256::from_le_bytes(new_two.block_hash().as_byte_array()),
+                    b'C',
+                    3
+                ),
+            ]
+        );
+        Ok(())
+    }
+
     #[test]
     fn a_disconnect_body_store_failure_moves_nothing() -> Result<(), Box<dyn std::error::Error>> {
         let ReorgBodyLoadingFixture {

crates/node/src/bitcoin_conf_compat.rs

@@ -68,10 +68,12 @@ fn apply_key(layer: &mut ConfigLayer, key: &str, value: &str) {
         "zmqpubhashtx" => push_endpoint(&mut layer.zmqpubhashtx, value),
         "zmqpubrawblock" => push_endpoint(&mut layer.zmqpubrawblock, value),
         "zmqpubrawtx" => push_endpoint(&mut layer.zmqpubrawtx, value),
+        "zmqpubsequence" => push_endpoint(&mut layer.zmqpubsequence, value),
         "zmqpubhashblockhwm" => layer.zmqpubhashblockhwm = value.parse().ok(),
         "zmqpubhashtxhwm" => layer.zmqpubhashtxhwm = value.parse().ok(),
         "zmqpubrawblockhwm" => layer.zmqpubrawblockhwm = value.parse().ok(),
         "zmqpubrawtxhwm" => layer.zmqpubrawtxhwm = value.parse().ok(),
+        "zmqpubsequencehwm" => layer.zmqpubsequencehwm = value.parse().ok(),
         _ => {}
     }
     if layer.rpc_user.is_some() || layer.rpc_password.is_some() {
@@ -196,6 +198,9 @@ impl ConfigLayerMerge for ConfigLayer {
         if other.zmqpubrawtx.is_some() {
             self.zmqpubrawtx.clone_from(&other.zmqpubrawtx);
         }
+        if other.zmqpubsequence.is_some() {
+            self.zmqpubsequence.clone_from(&other.zmqpubsequence);
+        }
         if other.zmqpubhashblockhwm.is_some() {
             self.zmqpubhashblockhwm = other.zmqpubhashblockhwm;
         }
@@ -208,5 +213,8 @@ impl ConfigLayerMerge for ConfigLayer {
         if other.zmqpubrawtxhwm.is_some() {
             self.zmqpubrawtxhwm = other.zmqpubrawtxhwm;
         }
+        if other.zmqpubsequencehwm.is_some() {
+            self.zmqpubsequencehwm = other.zmqpubsequencehwm;
+        }
     }
 }

crates/node/src/config.rs

@@ -151,6 +151,10 @@ pub struct Config {
     pub zmqpubrawblockhwm: Option<u32>,
     /// Optional `rawtx` PUB socket high-water mark.
     pub zmqpubrawtxhwm: Option<u32>,
+    /// ZMQ `sequence` PUB bind endpoints.
+    pub zmqpubsequence: Vec<String>,
+    /// Optional `sequence` PUB socket high-water mark.
+    pub zmqpubsequencehwm: Option<u32>,
     /// Block height at or below which script verification is skipped during block apply.
     ///
     /// On mainnet the default is the hash-pinned assume-valid anchor
@@ -211,6 +215,8 @@ impl fmt::Debug for Config {
             .field("zmqpubhashtxhwm", &self.zmqpubhashtxhwm)
             .field("zmqpubrawblockhwm", &self.zmqpubrawblockhwm)
             .field("zmqpubrawtxhwm", &self.zmqpubrawtxhwm)
+            .field("zmqpubsequence", &self.zmqpubsequence)
+            .field("zmqpubsequencehwm", &self.zmqpubsequencehwm)
             .field("assume_valid_height", &self.assume_valid_height)
             .finish_non_exhaustive()
     }
@@ -259,6 +265,8 @@ impl Config {
             zmqpubhashtxhwm: None,
             zmqpubrawblockhwm: None,
             zmqpubrawtxhwm: None,
+            zmqpubsequence: Vec::new(),
+            zmqpubsequencehwm: None,
             assume_valid_height: network
                 .assume_valid_anchor()
                 .map_or(0, |(height, _)| height),
@@ -357,6 +365,7 @@ impl Config {
             ("zmqpubhashtxhwm", self.zmqpubhashtxhwm),
             ("zmqpubrawblockhwm", self.zmqpubrawblockhwm),
             ("zmqpubrawtxhwm", self.zmqpubrawtxhwm),
+            ("zmqpubsequencehwm", self.zmqpubsequencehwm),
         ] {
             if hwm.is_some_and(|value| value > 2_147_483_647) {
                 bail!("{name} exceeds libzmq SNDHWM range");
@@ -393,6 +402,12 @@ impl Config {
             &self.zmqpubrawtx,
             self.zmqpubrawtxhwm,
         );
+        push_zmq_publications(
+            &mut publications,
+            crate::zmq_publisher::ZmqTopic::Sequence,
+            &self.zmqpubsequence,
+            self.zmqpubsequencehwm,
+        );
         publications
     }
 
@@ -427,6 +442,7 @@ impl Config {
         Ok(config)
     }
 
+    #[allow(clippy::too_many_lines)]
     fn apply_layer(&mut self, layer: &ConfigLayer) {
         if let Some(network) = layer.network {
             self.network = network;
@@ -513,6 +529,9 @@ impl Config {
         if let Some(endpoints) = &layer.zmqpubrawtx {
             self.zmqpubrawtx.clone_from(endpoints);
         }
+        if let Some(endpoints) = &layer.zmqpubsequence {
+            self.zmqpubsequence.clone_from(endpoints);
+        }
         if let Some(hwm) = layer.zmqpubhashblockhwm {
             self.zmqpubhashblockhwm = Some(hwm);
         }
@@ -525,6 +544,9 @@ impl Config {
         if let Some(hwm) = layer.zmqpubrawtxhwm {
             self.zmqpubrawtxhwm = Some(hwm);
         }
+        if let Some(hwm) = layer.zmqpubsequencehwm {
+            self.zmqpubsequencehwm = Some(hwm);
+        }
         if let Some(height) = layer.assume_valid_height {
             self.assume_valid_height = height;
         }
@@ -624,6 +646,8 @@ pub(crate) struct ConfigLayer {
     pub(crate) zmqpubrawblock: Option<Vec<String>>,
     #[arg(long = "zmqpubrawtx", value_delimiter = ',')]
     pub(crate) zmqpubrawtx: Option<Vec<String>>,
+    #[arg(long = "zmqpubsequence", value_delimiter = ',')]
+    pub(crate) zmqpubsequence: Option<Vec<String>>,
     #[arg(long = "zmqpubhashblockhwm")]
     pub(crate) zmqpubhashblockhwm: Option<u32>,
     #[arg(long = "zmqpubhashtxhwm")]
@@ -632,6 +656,8 @@ pub(crate) struct ConfigLayer {
     pub(crate) zmqpubrawblockhwm: Option<u32>,
     #[arg(long = "zmqpubrawtxhwm")]
     pub(crate) zmqpubrawtxhwm: Option<u32>,
+    #[arg(long = "zmqpubsequencehwm")]
+    pub(crate) zmqpubsequencehwm: Option<u32>,
     #[arg(long = "assume-valid-height")]
     pub(crate) assume_valid_height: Option<u32>,
 }
@@ -708,6 +734,9 @@ impl ConfigLayer {
                 "BITCOIN_RS_ZMQPUBRAWTX" => {
                     layer.zmqpubrawtx = Some(parse_string_list(value));
                 }
+                "BITCOIN_RS_ZMQPUBSEQUENCE" => {
+                    layer.zmqpubsequence = Some(parse_string_list(value));
+                }
                 "BITCOIN_RS_ZMQPUBHASHBLOCKHWM" => {
                     layer.zmqpubhashblockhwm = Some(value.parse()?);
                 }
@@ -720,6 +749,9 @@ impl ConfigLayer {
                 "BITCOIN_RS_ZMQPUBRAWTXHWM" => {
                     layer.zmqpubrawtxhwm = Some(value.parse()?);
                 }
+                "BITCOIN_RS_ZMQPUBSEQUENCEHWM" => {
+                    layer.zmqpubsequencehwm = Some(value.parse()?);
+                }
                 "BITCOIN_RS_ASSUME_VALID_HEIGHT" => {
                     layer.assume_valid_height = Some(value.parse()?);
                 }

crates/node/src/lib.rs

@@ -65,4 +65,6 @@ pub use run::run;
 pub use state::{ApplyError, DisconnectError};
 pub use sync::BlockSync;
 pub use utxo_view::UtxoSetView;
-pub use zmq_publisher::{NoOpZmqPublisher, SocketZmqPublisher, TracingZmqPublisher, ZmqPublisher};
+pub use zmq_publisher::{
+    NoOpZmqPublisher, SequenceEvent, SocketZmqPublisher, TracingZmqPublisher, ZmqPublisher,
+};