Harden masking fallback and frame readers after flow sync

Co-Authored-By: brekotis <93345790+brekotis@users.noreply.github.com>

src/proxy/middle_relay/d2c.rs

@@ -55,6 +55,37 @@ pub(super) fn classify_me_d2c_flush_reason(
     MeD2cFlushReason::QueueDrain
 }
 
+pub(super) fn me_d2c_flush_reason_requires_client_flush(reason: MeD2cFlushReason) -> bool {
+    !matches!(reason, MeD2cFlushReason::QueueDrain)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn queue_drain_is_not_a_physical_flush_trigger() {
+        assert!(!me_d2c_flush_reason_requires_client_flush(
+            MeD2cFlushReason::QueueDrain
+        ));
+        assert!(me_d2c_flush_reason_requires_client_flush(
+            MeD2cFlushReason::AckImmediate
+        ));
+        assert!(me_d2c_flush_reason_requires_client_flush(
+            MeD2cFlushReason::BatchFrames
+        ));
+        assert!(me_d2c_flush_reason_requires_client_flush(
+            MeD2cFlushReason::BatchBytes
+        ));
+        assert!(me_d2c_flush_reason_requires_client_flush(
+            MeD2cFlushReason::MaxDelay
+        ));
+        assert!(me_d2c_flush_reason_requires_client_flush(
+            MeD2cFlushReason::Close
+        ));
+    }
+}
+
 pub(super) fn observe_me_d2c_flush_event(
     stats: &Stats,
     reason: MeD2cFlushReason,

src/proxy/middle_relay/session.rs

@@ -491,12 +491,18 @@ where
                         d2c_flush_policy.max_bytes,
                         max_delay_fired,
                     );
-                    let flush_started_at = if stats_clone.telemetry_policy().me_level.allows_debug() {
+                    let physical_flush =
+                        me_d2c_flush_reason_requires_client_flush(flush_reason);
+                    let flush_started_at = if physical_flush
+                        && stats_clone.telemetry_policy().me_level.allows_debug()
+                    {
                         Some(Instant::now())
                     } else {
                         None
                     };
-                    flush_client_or_cancel(&mut writer, &flow_cancel_me_writer).await?;
+                    if physical_flush {
+                        flush_client_or_cancel(&mut writer, &flow_cancel_me_writer).await?;
+                    }
                     let flush_duration_us = flush_started_at.map(|started| {
                         started
                             .elapsed()