Relax the watermark rule in the runtime (#7188)

* Relax the watermark rule in the runtime

* make comment more clear

* add hrmp test

* remove TODO now comment

polkadot/runtime/parachains/src/hrmp/tests.rs

@@ -660,3 +660,48 @@ fn cancel_pending_open_channel_request() {
 		Hrmp::assert_storage_consistency_exhaustive();
 	});
 }
+
+#[test]
+fn watermark_maxed_out_at_relay_parent() {
+	let para_a = 32.into();
+	let para_b = 64.into();
+
+	let mut genesis = GenesisConfigBuilder::default();
+	genesis.hrmp_channel_max_message_size = 20;
+	genesis.hrmp_channel_max_total_size = 20;
+	new_test_ext(genesis.build()).execute_with(|| {
+		register_parachain(para_a);
+		register_parachain(para_b);
+
+		run_to_block(5, Some(vec![4, 5]));
+		Hrmp::init_open_channel(para_a, para_b, 2, 20).unwrap();
+		Hrmp::accept_open_channel(para_b, para_a).unwrap();
+
+		// On Block 6:
+		// A sends a message to B
+		run_to_block(6, Some(vec![6]));
+		assert!(channel_exists(para_a, para_b));
+		let msgs: HorizontalMessages =
+			vec![OutboundHrmpMessage { recipient: para_b, data: b"this is an emergency".to_vec() }]
+				.try_into()
+				.unwrap();
+		let config = Configuration::config();
+		assert!(Hrmp::check_outbound_hrmp(&config, para_a, &msgs).is_ok());
+		let _ = Hrmp::queue_outbound_hrmp(para_a, msgs);
+		Hrmp::assert_storage_consistency_exhaustive();
+
+		// On block 8:
+		// B receives the message sent by A. B sets the watermark to 7.
+		run_to_block(8, None);
+		assert!(Hrmp::check_hrmp_watermark(para_b, 7, 7).is_ok());
+		let _ = Hrmp::prune_hrmp(para_b, 7);
+		Hrmp::assert_storage_consistency_exhaustive();
+
+		// On block 9:
+		// B includes a candidate with the same relay parent as before.
+		run_to_block(9, None);
+		assert!(Hrmp::check_hrmp_watermark(para_b, 7, 7).is_ok());
+		let _ = Hrmp::prune_hrmp(para_b, 7);
+		Hrmp::assert_storage_consistency_exhaustive();
+	});
+}