nnrp_runtime/
pressure.rs

1use nnrp_core::{MessageType, NnrpError, PressureMetadata};
2
3#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
4pub struct RuntimePressureState {
5    pub outbound_credit_window: u64,
6    pub inbound_credit_window: u64,
7    pub remote_backpressure_level: u16,
8    pub remote_backpressure_reason: u16,
9    pub remote_retry_after_ms: u32,
10    pub remote_pressure_flags: u32,
11    pub local_backpressure_level: u16,
12    pub local_backpressure_reason: u16,
13    pub local_retry_after_ms: u32,
14    pub local_pressure_flags: u32,
15}
16
17impl RuntimePressureState {
18    pub fn apply_inbound(
19        &mut self,
20        message_type: MessageType,
21        metadata: PressureMetadata,
22    ) -> Result<(), NnrpError> {
23        match message_type {
24            MessageType::CreditUpdate => {
25                self.outbound_credit_window = metadata.credit_window;
26                Ok(())
27            }
28            MessageType::Backpressure => {
29                self.outbound_credit_window = metadata.credit_window;
30                self.remote_backpressure_level = metadata.pressure_level;
31                self.remote_backpressure_reason = metadata.pressure_reason;
32                self.remote_retry_after_ms = metadata.retry_after_ms;
33                self.remote_pressure_flags = metadata.flags;
34                Ok(())
35            }
36            _ => Err(NnrpError::InvalidProtocolCombination {
37                rule: "pressure state inbound update requires BACKPRESSURE or CREDIT_UPDATE",
38            }),
39        }
40    }
41
42    pub fn apply_outbound(
43        &mut self,
44        message_type: MessageType,
45        metadata: PressureMetadata,
46    ) -> Result<(), NnrpError> {
47        match message_type {
48            MessageType::CreditUpdate => {
49                self.inbound_credit_window = metadata.credit_window;
50                Ok(())
51            }
52            MessageType::Backpressure => {
53                self.inbound_credit_window = metadata.credit_window;
54                self.local_backpressure_level = metadata.pressure_level;
55                self.local_backpressure_reason = metadata.pressure_reason;
56                self.local_retry_after_ms = metadata.retry_after_ms;
57                self.local_pressure_flags = metadata.flags;
58                Ok(())
59            }
60            _ => Err(NnrpError::InvalidProtocolCombination {
61                rule: "pressure state outbound update requires BACKPRESSURE or CREDIT_UPDATE",
62            }),
63        }
64    }
65}
66
67#[cfg(test)]
68mod tests {
69    use nnrp_core::BackpressureLevel;
70
71    use super::*;
72
73    #[test]
74    fn inbound_pressure_updates_outbound_window_and_remote_pressure() {
75        let mut state = RuntimePressureState::default();
76        state
77            .apply_inbound(
78                MessageType::CreditUpdate,
79                PressureMetadata {
80                    scope_id: 1,
81                    credit_window: 8,
82                    pressure_level: BackpressureLevel::None as u16,
83                    pressure_reason: 0,
84                    retry_after_ms: 0,
85                    flags: 0,
86                },
87            )
88            .unwrap();
89        assert_eq!(state.outbound_credit_window, 8);
90
91        state
92            .apply_inbound(
93                MessageType::Backpressure,
94                PressureMetadata {
95                    scope_id: 1,
96                    credit_window: 3,
97                    pressure_level: BackpressureLevel::Hard as u16,
98                    pressure_reason: 11,
99                    retry_after_ms: 50,
100                    flags: 2,
101                },
102            )
103            .unwrap();
104        assert_eq!(state.outbound_credit_window, 3);
105        assert_eq!(
106            state.remote_backpressure_level,
107            BackpressureLevel::Hard as u16
108        );
109        assert_eq!(state.remote_backpressure_reason, 11);
110        assert_eq!(state.remote_retry_after_ms, 50);
111        assert_eq!(state.remote_pressure_flags, 2);
112    }
113
114    #[test]
115    fn outbound_pressure_updates_inbound_window_and_local_pressure() {
116        let mut state = RuntimePressureState::default();
117        state
118            .apply_outbound(
119                MessageType::CreditUpdate,
120                PressureMetadata {
121                    scope_id: 1,
122                    credit_window: 13,
123                    pressure_level: BackpressureLevel::None as u16,
124                    pressure_reason: 0,
125                    retry_after_ms: 0,
126                    flags: 0,
127                },
128            )
129            .unwrap();
130        assert_eq!(state.inbound_credit_window, 13);
131
132        state
133            .apply_outbound(
134                MessageType::Backpressure,
135                PressureMetadata {
136                    scope_id: 1,
137                    credit_window: 5,
138                    pressure_level: BackpressureLevel::Soft as u16,
139                    pressure_reason: 7,
140                    retry_after_ms: 25,
141                    flags: 4,
142                },
143            )
144            .unwrap();
145        assert_eq!(state.inbound_credit_window, 5);
146        assert_eq!(
147            state.local_backpressure_level,
148            BackpressureLevel::Soft as u16
149        );
150        assert_eq!(state.local_backpressure_reason, 7);
151        assert_eq!(state.local_retry_after_ms, 25);
152        assert_eq!(state.local_pressure_flags, 4);
153    }
154
155    #[test]
156    fn pressure_state_rejects_non_pressure_messages() {
157        let mut state = RuntimePressureState::default();
158        let metadata = PressureMetadata {
159            scope_id: 1,
160            credit_window: 1,
161            pressure_level: BackpressureLevel::None as u16,
162            pressure_reason: 0,
163            retry_after_ms: 0,
164            flags: 0,
165        };
166
167        assert!(matches!(
168            state.apply_inbound(MessageType::Cancel, metadata),
169            Err(NnrpError::InvalidProtocolCombination { .. })
170        ));
171        assert!(matches!(
172            state.apply_outbound(MessageType::Cancel, metadata),
173            Err(NnrpError::InvalidProtocolCombination { .. })
174        ));
175    }
176}
nnrp_runtime/pressure.rs

nnrp_runtime/
pressure.rs
