feat(link): gate type-scoped compaction behind V19/BlockLinks

The type-scoped compaction fix changes which messages a LinkCompactState
deletes, which changes the trie state root — so deploying it ungated would
fork a fixed node from the network on historical data. nindexer confirms the
risk is real: stray "unfollow"-typed links exist in history (a client bug;
"unfollow" squeaks past the <=8-byte link-type check) and ~25k follow
compact states exist, so a follow compaction that swept an unfollow link
under the old type-blind code would keep it under the fix.

Gate the behavior on a new EngineVersion::V19 / ProtocolFeature::BlockLinks:
Store::merge and merge_compact_state take scope_compaction_by_type, which the
engine derives from the message's version (is_enabled(BlockLinks)). Pre-V19
stays type-blind (preserves historical state roots); V19+ scopes deletion to
the compact state's own link type so a follow compaction can't delete block
links. The Store layer stays version-agnostic — it receives the bool, the
engine decides it.

Mainnet/testnet V19 activation is a placeholder pending rollout coordination;
devnet runs V19 from genesis. link_store tests pin both regimes (scoped keeps
other link types, type-blind deletes them).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

Author: bond00729

src/storage/store/account/cast_store_tests.rs

@@ -59,7 +59,7 @@ mod tests {
         err_message: String,
     ) {
         let mut txn = RocksDbTransactionBatch::new();
-        let result = store.merge(&message, &mut txn);
+        let result = store.merge(&message, &mut txn, false);
         assert!(result.is_err());
         let error = result.unwrap_err();
         assert_eq!(error.code, err_code);
@@ -73,7 +73,7 @@ mod tests {
         deleted_messages: Vec<message::Message>,
     ) {
         let mut txn = RocksDbTransactionBatch::new();
-        let result = store.merge(message, &mut txn).unwrap();
+        let result = store.merge(message, &mut txn, false).unwrap();
         assert_eq!(result.r#type(), HubEventType::MergeMessage);
         match &result.body {
             Some(hub_event::Body::MergeMessageBody(body)) => {
@@ -96,7 +96,7 @@ mod tests {
         let mut events = Vec::new();
 
         for message in messages {
-            let result = store.merge(message, &mut txn).unwrap();
+            let result = store.merge(message, &mut txn, false).unwrap();
             events.push(result.clone());
             assert_eq!(result.r#type(), HubEventType::MergeMessage);
             match &result.body {
@@ -1271,7 +1271,7 @@ mod tests {
         );
 
         let mut txn1 = RocksDbTransactionBatch::new();
-        let _result1 = store.merge(&cast_remove_earlier, &mut txn1).unwrap();
+        let _result1 = store.merge(&cast_remove_earlier, &mut txn1, false).unwrap();
         db.commit(txn1).unwrap();
 
         merge_message_failure(

src/storage/store/account/link_store_tests.rs

@@ -50,7 +50,7 @@ mod tests {
         message: &message::Message,
     ) {
         let mut txn = RocksDbTransactionBatch::new();
-        let result = store.merge(&message, &mut txn).unwrap();
+        let result = store.merge(&message, &mut txn, false).unwrap();
         assert_eq!(result.r#type(), HubEventType::MergeMessage);
         match &result.body {
             Some(hub_event::Body::MergeMessageBody(body)) => {
@@ -64,14 +64,41 @@ mod tests {
         db.commit(txn).unwrap();
     }
 
+    // Shared fixture for the compaction tests: a follow compact state at ts 2000 that
+    // references neither TARGET_FID nor any block target.
+    fn non_target_follow_compact_state() -> message::Message {
+        messages_factory::links::create_link_compact_state(
+            FID_FOR_TEST,
+            LINK_TYPE_FOLLOW,
+            vec![FID_FOR_TEST + 100],
+            Some(2000),
+            None,
+        )
+    }
+
+    // Merge a compact state at the given scope and commit. (merge_message_success can't be
+    // reused: compaction deletes messages, which that helper asserts does not happen.)
+    fn merge_compact_state_with_scope(
+        store: &Store<LinkStore>,
+        db: &Arc<RocksDB>,
+        message: &message::Message,
+        scope_compaction_by_type: bool,
+    ) {
+        let mut txn = RocksDbTransactionBatch::new();
+        store
+            .merge(message, &mut txn, scope_compaction_by_type)
+            .unwrap();
+        db.commit(txn).unwrap();
+    }
+
     fn merge_message_with_conflicts(
         store: &Store<LinkStore>,
         db: &Arc<RocksDB>,
         message: &message::Message,
         deleted_messages: Vec<message::Message>,
     ) {
         let mut txn = RocksDbTransactionBatch::new();
-        let result = store.merge(&message, &mut txn).unwrap();
+        let result = store.merge(&message, &mut txn, false).unwrap();
         assert_eq!(result.r#type(), HubEventType::MergeMessage);
         match &result.body {
             Some(hub_event::Body::MergeMessageBody(body)) => {
@@ -92,7 +119,7 @@ mod tests {
         err_message: &str,
     ) {
         let mut txn = RocksDbTransactionBatch::new();
-        let result = store.merge(&message, &mut txn);
+        let result = store.merge(&message, &mut txn, false);
         assert!(result.is_err());
         let err = result.err().unwrap();
         assert_eq!(err.code, err_code);
@@ -2279,14 +2306,8 @@ mod tests {
         merge_message_success(&store, &db, &block_add);
 
         // A later follow compact state that does not reference TARGET_FID at all.
-        let follow_compact_state = messages_factory::links::create_link_compact_state(
-            FID_FOR_TEST,
-            LINK_TYPE_FOLLOW,
-            vec![FID_FOR_TEST + 100],
-            Some(2000),
-            None,
-        );
-        merge_message_success(&store, &db, &follow_compact_state);
+        let follow_compact_state = non_target_follow_compact_state();
+        merge_compact_state_with_scope(&store, &db, &follow_compact_state, true);
 
         // The block link must still be active, because the follow compaction must
         // not touch links of a different type.
@@ -2319,25 +2340,113 @@ mod tests {
         merge_message_success(&store, &db, &block_remove);
 
         // A later follow compact state.
-        let follow_compact_state = messages_factory::links::create_link_compact_state(
+        let follow_compact_state = non_target_follow_compact_state();
+        merge_compact_state_with_scope(&store, &db, &follow_compact_state, true);
+
+        // The block remove tombstone must survive the follow compaction.
+        let retrieved = LinkStore::get_link_remove(
+            &store,
+            FID_FOR_TEST,
+            LINK_TYPE_BLOCK.to_string(),
+            Some(Target::TargetFid(TARGET_FID)),
+        )
+        .unwrap()
+        .unwrap();
+        assert_eq!(retrieved, block_remove);
+    }
+
+    #[test]
+    fn test_legacy_follow_compaction_deletes_other_link_type_adds() {
+        // Pre-V19 (type-blind) behavior: with scope_compaction_by_type = false, a "follow"
+        // compaction deletes a "block" add whose target is not in the compact state. This is
+        // the historical behavior preserved on replay, so a fixed node doesn't fork.
+        let (store, db, _temp_dir) = create_test_store();
+
+        let block_add = messages_factory::links::create_link_add(
+            FID_FOR_TEST,
+            LINK_TYPE_BLOCK,
+            TARGET_FID,
+            Some(1000),
+            None,
+        );
+        merge_message_success(&store, &db, &block_add);
+
+        let follow_compact_state = non_target_follow_compact_state();
+        merge_compact_state_with_scope(&store, &db, &follow_compact_state, false);
+
+        let retrieved = LinkStore::get_link_add(
+            &store,
+            FID_FOR_TEST,
+            LINK_TYPE_BLOCK.to_string(),
+            Some(Target::TargetFid(TARGET_FID)),
+        )
+        .unwrap();
+        assert!(retrieved.is_none());
+    }
+
+    #[test]
+    fn test_legacy_follow_compaction_deletes_other_link_type_removes() {
+        // Pre-V19 (type-blind) behavior: with scope_compaction_by_type = false, a "follow"
+        // compaction also deletes a "block" remove tombstone.
+        let (store, db, _temp_dir) = create_test_store();
+
+        let block_remove = messages_factory::links::create_link_remove(
+            FID_FOR_TEST,
+            LINK_TYPE_BLOCK,
+            TARGET_FID,
+            Some(1000),
+            None,
+        );
+        merge_message_success(&store, &db, &block_remove);
+
+        let follow_compact_state = non_target_follow_compact_state();
+        merge_compact_state_with_scope(&store, &db, &follow_compact_state, false);
+
+        let retrieved = LinkStore::get_link_remove(
+            &store,
+            FID_FOR_TEST,
+            LINK_TYPE_BLOCK.to_string(),
+            Some(Target::TargetFid(TARGET_FID)),
+        )
+        .unwrap();
+        assert!(retrieved.is_none());
+    }
+
+    #[test]
+    fn test_block_compaction_does_not_delete_follow_links() {
+        // The gate is symmetric: a "block" compact state must only compact "block" links and
+        // leave "follow" links intact (the direction blocks actually ship for).
+        let (store, db, _temp_dir) = create_test_store();
+
+        // A follow add, older than the block compact state, whose target is not listed.
+        let follow_add = messages_factory::links::create_link_add(
             FID_FOR_TEST,
             LINK_TYPE_FOLLOW,
+            TARGET_FID,
+            Some(1000),
+            None,
+        );
+        merge_message_success(&store, &db, &follow_add);
+
+        let block_compact_state = messages_factory::links::create_link_compact_state(
+            FID_FOR_TEST,
+            LINK_TYPE_BLOCK,
             vec![FID_FOR_TEST + 100],
             Some(2000),
             None,
         );
-        merge_message_success(&store, &db, &follow_compact_state);
+        merge_compact_state_with_scope(&store, &db, &block_compact_state, true);
 
-        // The block remove tombstone must survive the follow compaction.
-        let retrieved = LinkStore::get_link_remove(
+        // The follow link survives: a block compaction must not touch follow links.
+        let retrieved = LinkStore::get_link_add(
             &store,
             FID_FOR_TEST,
-            LINK_TYPE_BLOCK.to_string(),
+            LINK_TYPE_FOLLOW.to_string(),
             Some(Target::TargetFid(TARGET_FID)),
         )
         .unwrap()
         .unwrap();
-        assert_eq!(retrieved, block_remove);
+        assert_eq!(retrieved, follow_add);
     }
 
     #[test]
@@ -2369,19 +2478,11 @@ mod tests {
         merge_message_success(&store, &db, &follow_remove);
 
         // A later follow compact state that references neither target.
-        let follow_compact_state = messages_factory::links::create_link_compact_state(
-            FID_FOR_TEST,
-            LINK_TYPE_FOLLOW,
-            vec![FID_FOR_TEST + 100],
-            Some(2000),
-            None,
-        );
-        // Merge the compact state directly: it deletes the non-target follows, and we assert
-        // on the resulting store state below rather than on the (scan-order-dependent) list of
-        // deleted messages, which is an implementation detail rather than the contract here.
-        let mut txn = RocksDbTransactionBatch::new();
-        store.merge(&follow_compact_state, &mut txn).unwrap();
-        db.commit(txn).unwrap();
+        let follow_compact_state = non_target_follow_compact_state();
+        // The compaction deletes the non-target follows; we assert on the resulting store
+        // state below rather than on the (scan-order-dependent) list of deleted messages,
+        // which is an implementation detail rather than the contract here.
+        merge_compact_state_with_scope(&store, &db, &follow_compact_state, true);
 
         // The non-target follow add must be deleted by the compaction.
         let retrieved_add = LinkStore::get_link_add(

src/storage/store/account/reaction_store_tests.rs

@@ -31,7 +31,7 @@ mod tests {
         message: &message::Message,
     ) {
         let mut txn = RocksDbTransactionBatch::new();
-        let result = store.merge(&message, &mut txn).unwrap();
+        let result = store.merge(&message, &mut txn, false).unwrap();
         assert_eq!(result.r#type(), HubEventType::MergeMessage);
         match &result.body {
             Some(hub_event::Body::MergeMessageBody(body)) => {
@@ -52,7 +52,7 @@ mod tests {
         deleted_messages: Vec<message::Message>,
     ) {
         let mut txn = RocksDbTransactionBatch::new();
-        let result = store.merge(&message, &mut txn).unwrap();
+        let result = store.merge(&message, &mut txn, false).unwrap();
         assert_eq!(result.r#type(), HubEventType::MergeMessage);
         match &result.body {
             Some(hub_event::Body::MergeMessageBody(body)) => {
@@ -73,7 +73,7 @@ mod tests {
         err_message: &str,
     ) {
         let mut txn = RocksDbTransactionBatch::new();
-        let result = store.merge(&message, &mut txn);
+        let result = store.merge(&message, &mut txn, false);
         assert!(result.is_err());
         let error = result.unwrap_err();
         assert_eq!(error.code, err_code);

src/storage/store/account/storage_lend_store.rs

@@ -366,7 +366,7 @@ impl StorageLendStore {
         txn: &mut RocksDbTransactionBatch,
     ) -> Result<Vec<HubEvent>, HubError> {
         let mut events = vec![];
-        events.push(store.merge(message, txn)?);
+        events.push(store.merge(message, txn, false)?);
         match message.data.as_ref().unwrap().body.as_ref().unwrap() {
             proto::message_data::Body::LendStorageBody(lend_storage_body) => {
                 // Prune out the lend messages where storage is revoked so they don't consume storage.

src/storage/store/account/storage_lend_store_tests.rs

@@ -29,7 +29,7 @@ mod tests {
         message: &message::Message,
     ) {
         let mut txn = RocksDbTransactionBatch::new();
-        let result = store.merge(&message, &mut txn).unwrap();
+        let result = store.merge(&message, &mut txn, false).unwrap();
         assert_eq!(result.r#type(), HubEventType::MergeMessage);
         match &result.body {
             Some(hub_event::Body::MergeMessageBody(body)) => {
@@ -50,7 +50,7 @@ mod tests {
         deleted_messages: Vec<message::Message>,
     ) {
         let mut txn = RocksDbTransactionBatch::new();
-        let result = store.merge(&message, &mut txn).unwrap();
+        let result = store.merge(&message, &mut txn, false).unwrap();
         assert_eq!(result.r#type(), HubEventType::MergeMessage);
         match &result.body {
             Some(hub_event::Body::MergeMessageBody(body)) => {
@@ -71,7 +71,7 @@ mod tests {
         err_message: &str,
     ) {
         let mut txn = RocksDbTransactionBatch::new();
-        let result = store.merge(&message, &mut txn);
+        let result = store.merge(&message, &mut txn, false);
         assert!(result.is_err());
         let error = result.unwrap_err();
         assert_eq!(error.code, err_code);

src/storage/store/account/store.rs

@@ -564,6 +564,7 @@ impl<T: StoreDef + Clone> Store<T> {
         &self,
         message: &Message,
         txn: &mut RocksDbTransactionBatch,
+        scope_compaction_by_type: bool,
     ) -> Result<HubEvent, HubError> {
         if !self.store_def.is_add_type(message)
             && !(self.store_def.remove_type_supported() && self.store_def.is_remove_type(message))
@@ -579,7 +580,7 @@ impl<T: StoreDef + Clone> Store<T> {
         let ts_hash = make_ts_hash(message.data.as_ref().unwrap().timestamp, &message.hash)?;
 
         if self.store_def().is_compact_state_type(message) {
-            self.merge_compact_state(message, txn)
+            self.merge_compact_state(message, txn, scope_compaction_by_type)
         } else if self.store_def.is_add_type(message) {
             self.merge_add(&ts_hash, message, txn)
         } else {
@@ -649,6 +650,7 @@ impl<T: StoreDef + Clone> Store<T> {
         &self,
         message: &Message,
         txn: &mut RocksDbTransactionBatch,
+        scope_compaction_by_type: bool,
     ) -> Result<HubEvent, HubError> {
         let mut merge_conflicts = vec![];
 
@@ -682,9 +684,8 @@ impl<T: StoreDef + Clone> Store<T> {
         let (fid, compact_state_timestamp, target_fids, compact_state_type) =
             self.read_compact_state_details(message)?;
 
-        // Go over all the messages for this Fid of the SAME link type as the compact state
-        // (a "follow" compaction must not touch other-typed links, e.g. "block") that are
-        // older than the compact state message and
+        // Go over all messages for this Fid older than the compact state (when type-scoped,
+        // only those of the compact state's own link type — see below) and
         // 1. Delete all remove messages
         // 2. Delete all add messages that are not in the target_fids list
         let prefix = &make_message_primary_key(fid, self.store_def.postfix(), None);
@@ -701,12 +702,14 @@ impl<T: StoreDef + Clone> Store<T> {
                     return Ok(true);
                 }
 
-                // Only compact links of the same type as the compact state message.
-                // A "follow" compaction must not delete other-typed links, e.g. "block".
                 if let Some(Body::LinkBody(link_body)) =
                     message.data.as_ref().and_then(|data| data.body.as_ref())
                 {
-                    if link_body.r#type == compact_state_type {
+                    // When scope_compaction_by_type is set, only compact links of the compact
+                    // state's own type (so a "follow" compaction can't delete "block" links);
+                    // otherwise compaction is type-blind. The caller sets this from the engine
+                    // version (ProtocolFeature::BlockLinks).
+                    if !scope_compaction_by_type || link_body.r#type == compact_state_type {
                         if self.store_def.is_remove_type(&message) {
                             merge_conflicts.push(message);
                         } else if self.store_def.is_add_type(&message) {