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
@@ -2330,6 +2340,11 @@ fn apply_block_admitted(
         }
     }
     handles.applied_tip.store(Some(Arc::new(tip.clone())));
+    if handles.zmq_publisher.wants_notifications() {
+        handles
+            .zmq_publisher
+            .publish_sequence(crate::zmq_publisher::SequenceEvent::Connected(tip.hash));
+    }
     if let Some(sampler) = &handles.g2_muhash_sampler
         && sampler.wants_height(height)
     {
@@ -8678,6 +8693,183 @@ 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(())
+    }
+
+    #[derive(Debug)]
+    struct AppliedTipVisiblePublisher {
+        applied_tip: Arc<ArcSwapOption<TipSnapshot>>,
+        expected: Hash256,
+        seen: Mutex<Vec<Hash256>>,
+    }
+
+    impl crate::ZmqPublisher for AppliedTipVisiblePublisher {
+        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) {
+            if let crate::SequenceEvent::Connected(hash) = event {
+                assert_eq!(
+                    self.applied_tip.load_full().as_deref().map(|tip| tip.hash),
+                    Some(self.expected),
+                    "applied tip must be visible before publishing C"
+                );
+                self.seen.lock().push(hash);
+            }
+        }
+    }
+
+    #[test]
+    fn connected_sequence_event_observes_the_published_applied_tip()
+    -> Result<(), Box<dyn std::error::Error>> {
+        let handles = apply_handles_without_tx_index(Network::Regtest, Arc::new(UtxoSet::new()));
+        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 block = mined_block_with_prev_hash_and_transactions(
+            genesis.block_hash(),
+            vec![coinbase_transaction(5)],
+        )?;
+        let expected = Hash256::from_le_bytes(block.block_hash().as_byte_array());
+        let publisher = Arc::new(AppliedTipVisiblePublisher {
+            applied_tip: Arc::clone(&handles.applied_tip),
+            expected,
+            seen: Mutex::new(Vec::new()),
+        });
+        let publisher_handle: Arc<dyn crate::ZmqPublisher> = publisher.clone();
+        let handles = handles.with_zmq_publisher(publisher_handle);
+
+        apply_block(&handles, &block)?;
+
+        assert_eq!(*publisher.seen.lock(), vec![expected]);
+        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;
+        }
     }
 }