rzmq 0.5.25

#![allow(dead_code, unused_variables, unused_mut)]

use crate::error::ZmqError;
use crate::message::FrameBatch;
use crate::protocol::zmtp::actions::{AppAction, EngineOutput, NetAction};
use crate::protocol::zmtp::engine::{ZmtpEngine, ZmtpPhase};
use crate::protocol::zmtp::greeting::GREETING_LENGTH;
use crate::runtime::{ActorDropGuard, ActorType, Command, ReusableBoxFuture, SystemEvent};
use crate::sessionx::regulator::SessionRegulator;
use crate::socket::ISocket;
use crate::socket::events::{SocketEvent, clean_endpoint_uri};
use crate::socket::options::ZmtpEngineConfig;
use crate::socket::patterns::ready_pipe_queue::PipeMessageSender;
use crate::throttle::AdaptiveThrottle;
use crate::transport::{ZmtpStdStream, ZmtpWriteHalf};
use crate::{Blob, MailboxReceiver, counter, log_carryover_drain, log_delivery_gap, log_gating_failure, log_session_diagnostics};

use super::message_processor::ZmqMessageProcessor;

use bytes::BytesMut;
use futures::FutureExt;
use std::fmt::Debug;
use std::sync::Arc;
use std::time::Duration;
use tokio::io::{AsyncReadExt, AsyncWriteExt};
use tokio::sync::{OwnedSemaphorePermit, broadcast};
use tokio::task::{JoinHandle, yield_now};
use tokio::time::{Instant as TokioInstant, MissedTickBehavior};

use super::egress_buffer::EgressBuffer;
use super::egress_driver::EgressDriver;
use super::pipe_manager::CorePipeManagerX;
use super::states::ActorConfigX;
use super::types::ConnectionPhaseX;

use std::sync::atomic::Ordering;

pub(crate) struct SessionConnectionActorX<S: ZmtpStdStream> {
  handle: usize,
  current_phase: ConnectionPhaseX,
  parent_socket_id: usize,
  actor_config: ActorConfigX,

  zmtp_engine: ZmtpEngine,
  read_half: Option<S::ReadHalf>,
  write_half: Option<S::WriteHalf>,
  /// Temporary read buffer used only during the handshake loop.
  handshake_read_buf: BytesMut,
  /// Data messages decoded during the handshake pass (e.g. a SUB's SUBSCRIBE
  /// frame that arrived bundled with the peer's final handshake bytes). These
  /// must be replayed into the operational ingress rather than dropped, or the
  /// early message is lost. See the operational-loop drain.
  pending_handshake_ingress: Vec<FrameBatch>,

  core_pipe_manager: CorePipeManagerX,

  command_mailbox_receiver: MailboxReceiver,
  system_event_receiver: broadcast::Receiver<SystemEvent>,

  pending_peer_identity_from_handshake: Option<Blob>,
  pending_peer_socket_type: Option<String>,
  error_for_drop_guard: Option<ZmqError>,

  ping_check_timer: Option<tokio::time::Interval>,

  handshake_deadline: Option<TokioInstant>,
  socket_logic: Arc<dyn ISocket>,
  session_regulator: SessionRegulator,
  _connection_permit: Option<OwnedSemaphorePermit>,
  /// Arc so the persistent ingress send future (`ingress_send_fut`) can own a
  /// clone; the actor never sends on the pipe while that future is active, so
  /// the pipe's single-producer contract is preserved.
  incoming_pipe_sender: Option<Arc<PipeMessageSender>>,
  /// Ingress read strategy, resolved once from the ingress kind when pipes
  /// attach: `true` (frame-in-place, no accumulator copy) for addressed sockets
  /// (REP/DEALER/ROUTER), `false` (copy into accumulator) for anonymous ones
  /// (PULL/SUB). See `ZmqMessageProcessor::read_and_process`.
  ingress_frame_in_place: bool,
  is_currently_congested: bool,

  /// LINGER-compliant drain: set when a *clean* shutdown trigger (Stop command,
  /// SocketClosing/ContextTerminating broadcast, mailbox closed) arrives while
  /// Operational. Instead of exiting the loop immediately — which drops queued
  /// egress (core pipe, carryover, egress buffer) on the floor — the loop keeps
  /// running with only the drain-relevant arms making progress, and transitions
  /// to ShuttingDownStream once all outbound queues are empty or the linger
  /// deadline expires.
  graceful_drain_requested: bool,
  /// Absolute deadline for the graceful drain, derived from the LINGER option
  /// at trigger time. `None` while draining = infinite linger (ZMQ -1).
  graceful_drain_deadline: Option<TokioInstant>,
  /// The command mailbox returned Err (all senders dropped). Latched so the
  /// select arm can be disabled — otherwise a closed mailbox is always ready
  /// and the biased select would spin on it during a graceful drain, starving
  /// the write arm.
  command_mailbox_closed: bool,

  #[cfg(target_os = "linux")]
  cork_info: Option<crate::sessionx::cork::TcpCorkInfoX>,
  #[cfg(not(target_os = "linux"))]
  cork_info: Option<()>,
}

impl<S> SessionConnectionActorX<S>
where
  S: ZmtpStdStream + Debug + Unpin + Send + 'static,
{
  pub(crate) fn create_and_spawn(
    handle: usize,
    parent_socket_id: usize,
    stream: S,
    actor_config: ActorConfigX,
    engine_config: Arc<ZmtpEngineConfig>,
    command_mailbox_receiver: MailboxReceiver,
    socket_logic: Arc<dyn ISocket>,
    connection_permit: Option<OwnedSemaphorePermit>,
  ) -> JoinHandle<()> {
    // Capture the raw fd for cork setup BEFORE consuming the stream via into_split.
    #[cfg(target_os = "linux")]
    let cork_info = {
      use std::os::fd::AsRawFd;
      if engine_config.use_cork {
        crate::sessionx::cork::try_create_cork_info(Some(&stream), true)
      } else {
        None
      }
    };
    #[cfg(not(target_os = "linux"))]
    let cork_info: Option<()> = None;

    // Split the stream into independent owned halves.
    let (read_half, write_half) = stream.into_split();

    let zmtp_engine = ZmtpEngine::new(actor_config.is_server_role, engine_config.clone());

    let mut ping_check_timer = None;
    if let Some(ivl) = engine_config.heartbeat_ivl {
      if !ivl.is_zero() {
        let mut timer = tokio::time::interval_at(TokioInstant::now() + ivl, ivl);
        timer.set_missed_tick_behavior(MissedTickBehavior::Delay);
        ping_check_timer = Some(timer);
      }
    }

    let handshake_deadline = engine_config
      .handshake_timeout
      .map(|d| TokioInstant::now() + d);

    let system_event_receiver = actor_config.context.event_bus().subscribe();

    let rcv_timeout = engine_config
      .rcvtimeo
      .unwrap_or(std::time::Duration::from_secs(30));
    let regulator_min_lifespan = if rcv_timeout < std::time::Duration::from_millis(1000) {
      std::time::Duration::from_millis(1000)
    } else {
      std::time::Duration::from_millis(1000)
    };

    let actor = Self {
      handle,
      current_phase: ConnectionPhaseX::HandshakeInProgress,
      parent_socket_id,
      actor_config,
      zmtp_engine,
      read_half: Some(read_half),
      write_half: Some(write_half),
      handshake_read_buf: BytesMut::with_capacity(GREETING_LENGTH * 4),
      pending_handshake_ingress: Vec::new(),
      core_pipe_manager: CorePipeManagerX::new(),
      command_mailbox_receiver,
      system_event_receiver,
      pending_peer_identity_from_handshake: None,
      pending_peer_socket_type: None,
      error_for_drop_guard: None,
      ping_check_timer,
      handshake_deadline,
      socket_logic,
      session_regulator: SessionRegulator::new(regulator_min_lifespan),
      _connection_permit: connection_permit,
      incoming_pipe_sender: None,
      ingress_frame_in_place: false,
      is_currently_congested: false,
      graceful_drain_requested: false,
      graceful_drain_deadline: None,
      command_mailbox_closed: false,
      cork_info,
    };

    let task_handle = tokio::spawn(actor.run_loop());
    task_handle
  }

  async fn run_loop(mut self) {
    tracing::info!(
      sca_handle = self.handle,
      uri = %self.actor_config.connected_endpoint_uri,
      role = if self.actor_config.is_server_role { "Server" } else { "Client" },
      "SessionConnectionActorX started."
    );

    let mut actor_drop_guard = ActorDropGuard::new(
      self.actor_config.context.clone(),
      self.handle,
      ActorType::Session,
      Some(self.actor_config.connected_endpoint_uri.clone()),
      Some(self.parent_socket_id),
    );

    let adaptive_throttle = {
      let mut config = self.zmtp_engine.config().throttle_config.clone();
      if self.actor_config.is_server_role {
        config.priority = crate::throttle::types::Priority::Egress;
      } else {
        config.priority = crate::throttle::types::Priority::Ingress;
      }
      AdaptiveThrottle::new(config)
    };

    let mut outgoing_batch: Vec<FrameBatch> =
      Vec::with_capacity(self.zmtp_engine.config().sndbatch_count);

    let mut egress_buffer = EgressBuffer::new();

    // Local FIFO queue decouples network read from application pipe send to prevent deadlock.
    let mut ingress_buffer: Vec<FrameBatch> =
      Vec::with_capacity(self.zmtp_engine.config().rcvbatch_count);

    // Send the initial greeting (client only; server returns empty output).
    let initial_out = self.zmtp_engine.start();
    self.apply_engine_output_handshake(initial_out).await;
    if matches!(self.current_phase, ConnectionPhaseX::Terminating) {
      self.finalize(actor_drop_guard).await;
      return;
    }

    // ── HANDSHAKE LOOP ────────────────────────────────────────────────────────
    // Synchronous: drive the engine to ZmtpPhase::Data before processing any
    // commands. This preserves the old behaviour where ScaInitializePipes stays
    // in the mailbox until the handshake is fully done.
    {
      let mut hs_read_half = self
        .read_half
        .take()
        .expect("read_half present at handshake start");

      let hs_timeout = self
        .zmtp_engine
        .config()
        .handshake_timeout
        .unwrap_or(Duration::from_secs(15));

      'handshake: loop {
        if self.zmtp_engine.phase == ZmtpPhase::Data
          || self.zmtp_engine.phase == ZmtpPhase::Closed
          || matches!(self.current_phase, ConnectionPhaseX::Terminating)
        {
          break 'handshake;
        }

        let read_result = tokio::time::timeout(
          hs_timeout,
          hs_read_half.read_buf(&mut self.handshake_read_buf),
        )
        .await;

        match read_result {
          Err(_elapsed) => {
            tracing::warn!(
              sca_handle = self.handle,
              uri = %self.actor_config.connected_endpoint_uri,
              "Overall ZMTP handshake timed out."
            );
            self.set_fatal_error(ZmqError::Timeout).await;
            break 'handshake;
          }
          Ok(Err(e)) => {
            self
              .set_fatal_error(ZmqError::from_io_endpoint(e, "handshake read"))
              .await;
            break 'handshake;
          }
          Ok(Ok(0)) => {
            self.set_fatal_error(ZmqError::ConnectionClosed).await;
            break 'handshake;
          }
          Ok(Ok(_)) => {
            let data = self.handshake_read_buf.split().freeze();
            let out = self.zmtp_engine.on_network_bytes(data);
            self.apply_engine_output_handshake(out).await;
          }
        }
      }

      self.read_half = Some(hs_read_half);
    }

    // Handshake done — wait for ScaInitializePipes before entering Operational.
    // ScaInitializePipes may already be in the mailbox (sent by SocketCore as soon
    // as it learned about the connection) or may arrive shortly after.
    if self.zmtp_engine.phase == ZmtpPhase::Data
      && !matches!(self.current_phase, ConnectionPhaseX::Terminating)
    {
      self.current_phase = ConnectionPhaseX::WaitingForPipes;
      while self.current_phase == ConnectionPhaseX::WaitingForPipes {
        match self.command_mailbox_receiver.recv().await {
          Ok(cmd) => self.process_command(cmd).await,
          Err(_) => {
            tracing::info!(
              sca_handle = self.handle,
              "SCA Command mailbox closed while waiting for pipes."
            );
            self.transition_to_shutdown_stream(None).await;
            break;
          }
        }
      }
    }

    // ── OPERATIONAL LOOP ──────────────────────────────────────────────────────
    if self.current_phase == ConnectionPhaseX::Operational {
      let mut message_processor = ZmqMessageProcessor::new();

      // Replay any data messages that were decoded during the handshake pass
      // (before the ingress pipe was active) so they reach the socket first and
      // in order — e.g. a SUB's SUBSCRIBE bundled with its final handshake bytes.
      if !self.pending_handshake_ingress.is_empty() {
        ingress_buffer.append(&mut self.pending_handshake_ingress);
      }

      let mut read_half = self
        .read_half
        .take()
        .expect("read_half must be present at operational start");
      let mut write_half = self
        .write_half
        .take()
        .expect("write_half must be present at operational start");

      let sndhwm = self.zmtp_engine.config().sndhwm.max(1);
      let sndbatch_count = self.zmtp_engine.config().sndbatch_count;

      let mut core_carryover: std::collections::VecDeque<FrameBatch> =
        std::collections::VecDeque::new();

      #[cfg(feature = "diagnostics")]
      let mut last_log_ms = 0u64;

      // Persistent ingress→pipe send. Owns the drained ingress buffer plus an
      // Arc'd sender and returns the buffer on completion so its allocation is
      // reused. Unlike an inline select! arm it is NOT dropped when another arm
      // wins — under backpressure the blocked send resumes where it left off
      // instead of re-running the try_send fast path (and its reserved_count
      // add/rollback churn) every loop iteration. The box itself is allocated
      // once and reused via ReusableBoxFuture::set().
      let mut ingress_send_fut: ReusableBoxFuture<
        'static,
        (Vec<FrameBatch>, Result<usize, ZmqError>),
      > = ReusableBoxFuture::new(async { (Vec::new(), Ok(0)) });
      let mut ingress_send_fut_active = false;

      'operational: while self.current_phase == ConnectionPhaseX::Operational {
        // Arm the persistent ingress send with whatever the read arm buffered.
        if !ingress_send_fut_active && !ingress_buffer.is_empty() {
          match self.incoming_pipe_sender.as_ref() {
            Some(sender) => {
              let sender = Arc::clone(sender);
              let mut buf = std::mem::take(&mut ingress_buffer);
              ingress_send_fut.set(async move {
                let res = sender.send_batch_mut(&mut buf).await;
                (buf, res)
              });
              ingress_send_fut_active = true;
            }
            None => {
              // No pipe sender (PUB/PUSH send-only) — silently discard.
              ingress_buffer.clear();
            }
          }
        }

        if !core_carryover.is_empty()
          && self.current_phase == ConnectionPhaseX::Operational
          && egress_buffer.pending_messages() < sndhwm
        {
          log_carryover_drain!(self.handle, core_carryover.len());
          outgoing_batch.clear();
          let pending = egress_buffer.pending_messages();
          self.fill_outgoing_batch(&mut outgoing_batch, &mut core_carryover, pending, sndhwm);
          self
            .stage_outgoing_batch(&outgoing_batch, &mut egress_buffer, sndhwm, &adaptive_throttle)
            .await;
        }

        // Graceful-drain completion check (LINGER path): all outbound queues
        // empty means everything the app queued has been handed to the kernel —
        // safe to FIN now.
        if self.graceful_drain_requested
          && self.core_pipe_manager.len() == 0
          && core_carryover.is_empty()
          && egress_buffer.is_empty()
        {
          self.graceful_drain_requested = false;
          self.graceful_drain_deadline = None;
          tracing::info!(
            sca_handle = self.handle,
            "SCA graceful drain complete — transitioning to ShuttingDownStream."
          );
          self.current_phase = ConnectionPhaseX::ShuttingDownStream;
          continue 'operational; // while-condition exits the loop
        }

        // Linger deadline for the graceful drain; pending() when not draining
        // or when linger is infinite.
        let mut drain_deadline_future = futures::future::pending().left_future();
        if self.graceful_drain_requested {
          if let Some(deadline) = self.graceful_drain_deadline {
            drain_deadline_future = tokio::time::sleep_until(deadline).right_future();
          }
        }

        let mut pong_timeout_future = futures::future::pending().left_future();
        if self.zmtp_engine.is_waiting_for_pong() {
          if let Some(deadline_std) = self.zmtp_engine.get_pong_deadline() {
            let deadline_tokio = TokioInstant::from_std(deadline_std);
            pong_timeout_future = tokio::time::sleep_until(deadline_tokio).right_future();
          }
        }

        tokio::select! {
          biased;

          maybe_cmd = self.command_mailbox_receiver.recv(), if !self.command_mailbox_closed => {
            match maybe_cmd {
              Ok(command) => self.process_command(command).await,
              Err(_) => {
                tracing::info!(sca_handle = self.handle, "SCA Command mailbox closed.");
                // Latch: a closed mailbox is always-ready; without disabling this
                // arm the biased select would spin here during a graceful drain.
                self.command_mailbox_closed = true;
                self.transition_to_shutdown_stream(None).await;
              }
            }
          }

          maybe_event = self.system_event_receiver.recv() => {
            match maybe_event {
              Ok(event) => self.process_system_event(event).await,
              Err(broadcast::error::RecvError::Lagged(n)) => {
                self.set_fatal_error(ZmqError::Internal(format!("System event lagged by {}", n))).await;
              }
              Err(broadcast::error::RecvError::Closed) => {
                self.set_fatal_error(ZmqError::Internal("System event channel closed".into())).await;
              }
            }
          }

          _ = async { self.ping_check_timer.as_mut().map_or(futures::future::pending().left_future(), |t| t.tick().right_future()).await },
            if self.ping_check_timer.is_some() => {
            let ping_out = self.zmtp_engine.on_tick(std::time::Instant::now());
            for action in ping_out.net_actions {
              if let NetAction::Send { data, .. } = action {
                egress_buffer.push_priority(data);
              }
            }
            for action in ping_out.app_actions {
              if let AppAction::PeerError(e) = action {
                self.set_fatal_error(e).await;
              }
            }
          }

          _ = pong_timeout_future, if self.zmtp_engine.is_waiting_for_pong() => {
            self.set_fatal_error(ZmqError::Timeout).await;
          }

          // LINGER deadline expired mid-drain: give up and FIN. Real message
          // loss (matching ZMQ linger semantics) — always logged.
          _ = drain_deadline_future, if self.graceful_drain_requested && self.graceful_drain_deadline.is_some() => {
            tracing::warn!(
              sca_handle = self.handle,
              egress_pending = egress_buffer.pending_messages(),
              core_carryover = core_carryover.len(),
              core_pipe_queued = self.core_pipe_manager.len(),
              "SCA: LINGER deadline expired during graceful drain — discarding remaining egress"
            );
            self.graceful_drain_requested = false;
            self.graceful_drain_deadline = None;
            self.current_phase = ConnectionPhaseX::ShuttingDownStream;
          }

          // Drain ingress buffer: decouples network read from application pipe send.
          // Persistent across iterations (armed at the top of the loop); losing
          // the select! race does not restart the blocked send.
          (returned_buf, write_res) = &mut ingress_send_fut, if ingress_send_fut_active => {
            ingress_send_fut_active = false;
            // Reclaim the (drained) buffer allocation for the read arm.
            ingress_buffer = returned_buf;
            match write_res {
              Ok(sent) => {
                if sent > 0 {
                  let throttle_guard = adaptive_throttle.begin_work_bulk(
                    crate::throttle::Direction::Ingress,
                    sent as u32,
                  );
                  if throttle_guard.should_throttle() {
                    yield_now().await;
                  }
                }
              }
              Err(e) => {
                tracing::debug!(sca_handle = self.handle, error = %e, "Ingress pipe closed during batch send.");
                self.transition_to_shutdown_stream(Some(e)).await;
              }
            }
          }

          // Ingress network read: gated on the buffer AND the in-flight pipe
          // send (the buffer is empty-by-move while the send future is active)
          // to propagate TCP backpressure.
          ingress_res = message_processor.read_and_process(&mut read_half, &mut self.zmtp_engine, self.ingress_frame_in_place),
            if ingress_buffer.is_empty() && !ingress_send_fut_active => {
            match ingress_res {
              Ok(engine_out) => {
                // Net actions: PONG frames and any other protocol-level sends.
                for action in engine_out.net_actions {
                  match action {
                    NetAction::Send { data, .. } => {
                      egress_buffer.push_priority(data);
                    }
                    NetAction::SetCork(enable) => {
                      #[cfg(target_os = "linux")]
                      if let Some(ci) = self.cork_info.as_mut() {
                        ci.apply_cork_state(enable, self.handle).await;
                      }
                    }
                    NetAction::ScheduleClose(_) => {
                      self.transition_to_shutdown_stream(None).await;
                    }
                  }
                }
                // App actions: fully-assembled data messages only (commands handled by engine).
                for action in engine_out.app_actions {
                  match action {
                    AppAction::DeliverMessage(batch) => {
                      ingress_buffer.push(batch);
                    }
                    AppAction::PeerError(e) => {
                      self.set_fatal_error(e).await;
                    }
                    AppAction::HandshakeComplete { .. } => {}
                  }
                }
              }
              Err(ZmqError::ConnectionClosed) => {
                tracing::info!(sca_handle = self.handle, "Peer closed connection.");
                self.transition_to_shutdown_stream(Some(ZmqError::ConnectionClosed)).await;
              }
              Err(e) => self.set_fatal_error(e).await,
            }
          }

          // Write arm: cancel-safe drain of the egress buffer.
          write_res = EgressDriver::new(&mut write_half, &mut egress_buffer, sndbatch_count, self.handle),
            if !egress_buffer.is_empty() => {
            match write_res {
              Ok(()) => {
                self.zmtp_engine.record_activity();
                if self.is_currently_congested
                  && egress_buffer.pending_messages() < sndhwm
                {
                  self.is_currently_congested = false;
                  if let Some(ref tx) = self.actor_config.monitor_tx {
                    let ep = clean_endpoint_uri(
                      &self.actor_config.logical_target_endpoint_uri,
                    ).to_owned();
                    let _ = tx.try_send(SocketEvent::ConnectionUncongested {
                      endpoint: ep,
                    });
                  }
                }
              }
              Err(e) => {
                self.set_fatal_error(e).await;
              }
            }
          }

          // Outgoing from SocketCore.
          maybe_msgs_from_core = async { self.core_pipe_manager.recv_from_core().await },
            if self.current_phase == ConnectionPhaseX::Operational
              && self.core_pipe_manager.is_attached()
              && core_carryover.is_empty()
              && egress_buffer.pending_messages() < sndhwm => {
            log_gating_failure!(egress_buffer.pending_messages(), sndhwm);
            match maybe_msgs_from_core {
              Ok(first_msgs) => {
                outgoing_batch.clear();
                outgoing_batch.push(first_msgs);
                let pending = egress_buffer.pending_messages();
                self.fill_outgoing_batch(&mut outgoing_batch, &mut core_carryover, pending, sndhwm);
                self
                  .stage_outgoing_batch(&outgoing_batch, &mut egress_buffer, sndhwm, &adaptive_throttle)
                  .await;
              }
              Err(_) => {
                tracing::info!(sca_handle = self.handle, "Pipe from SocketCore closed.");
                self.transition_to_shutdown_stream(Some(ZmqError::ConnectionClosed)).await;
              }
            }
          }

        }

        log_session_diagnostics!(last_log_ms, self, ingress_buffer, egress_buffer, sndhwm, core_carryover);
      }

      // Message-conservation audit: anything still queued when the operational
      // loop exits dies with this actor unless a later phase drains it. One log
      // line per connection at shutdown — egress loss becomes provable from the
      // log instead of inferred.
      #[cfg(feature = "diagnostics")]
      {
        let residual_egress = egress_buffer.pending_messages();
        let residual_carryover = core_carryover.len();
        let residual_core_pipe = self.core_pipe_manager.len();
        let residual_ingress = ingress_buffer.len();
        if residual_egress + residual_carryover + residual_core_pipe + residual_ingress > 0 {
          tracing::warn!(
            sca_handle = self.handle,
            uri = %self.actor_config.connected_endpoint_uri,
            phase = ?self.current_phase,
            egress_pending = residual_egress,
            core_carryover = residual_carryover,
            core_pipe_queued = residual_core_pipe,
            ingress_pending = residual_ingress,
            "[MSG-CONSERVATION] SCA exiting operational loop with undrained queues"
          );
        } else {
          tracing::debug!(
            sca_handle = self.handle,
            phase = ?self.current_phase,
            "[MSG-CONSERVATION] SCA exiting operational loop clean (all queues empty)"
          );
        }
      }

      self.read_half = Some(read_half);
      self.write_half = Some(write_half);

      if self.current_phase == ConnectionPhaseX::Terminating {
        log_delivery_gap!(
          self.handle,
          self.actor_config.connected_endpoint_uri,
          self.error_for_drop_guard,
          egress_buffer,
          core_carryover,
          ingress_buffer,
          outgoing_batch
        );
      }
    }

    if self.current_phase == ConnectionPhaseX::ShuttingDownStream {
      self.perform_graceful_shutdown().await;
    }

    tracing::info!(
      sca_handle = self.handle,
      "SCA loop finished. Final phase: {:?}.",
      self.current_phase
    );

    if self.current_phase != ConnectionPhaseX::Terminating {
      self.perform_final_cleanup_actions().await;
    }

    self.finalize(actor_drop_guard).await;
  }

  async fn finalize(mut self, mut actor_drop_guard: ActorDropGuard) {
    let had_error = self.error_for_drop_guard.is_some();

    if let Some(err) = self.error_for_drop_guard.take() {
      actor_drop_guard.set_error(err);
    } else {
      actor_drop_guard.waive();
    }

    drop(actor_drop_guard);

    if had_error {
      self.session_regulator.enforce_min_lifespan().await;
    }

    tracing::info!(
      sca_handle = self.handle,
      "SCA {} fully stopped.",
      self.actor_config.connected_endpoint_uri
    );
  }

  // --- Helper Methods ---

  async fn process_command(&mut self, command: Command) {
    tracing::debug!(
      sca_handle = self.handle,
      "Processing ScaCommand: {:?}",
      command
    );
    match command {
      Command::ScaInitializePipes {
        sca_handle_id,
        rx_from_core,
        core_pipe_read_id_for_incoming_routing,
        incoming_pipe_sender,
      } => {
        if self.handle != sca_handle_id {
          tracing::warn!(
            sca_handle = self.handle,
            expected_id = sca_handle_id,
            "Received ScaInitializePipes for wrong actor. Ignoring."
          );
          drop(rx_from_core);
          return;
        }
        if self.core_pipe_manager.is_attached() {
          tracing::warn!(
            sca_handle = self.handle,
            "Already attached. Ignoring ScaInitializePipes."
          );
          drop(rx_from_core);
          return;
        }

        if matches!(
          self.current_phase,
          ConnectionPhaseX::ShuttingDownStream | ConnectionPhaseX::Terminating
        ) {
          tracing::warn!(
            sca_handle = self.handle,
            phase = ?self.current_phase,
            "Received ScaInitializePipes while shutting down. Dropping pipes."
          );
          drop(rx_from_core);
          return;
        }

        self.incoming_pipe_sender = incoming_pipe_sender.map(Arc::new);
        // Resolve the ingress read strategy once: addressed sockets
        // (REP/DEALER/ROUTER) frame in place; anonymous ones (PULL/SUB) and the
        // PUB subscription sink copy into the accumulator.
        self.ingress_frame_in_place = matches!(
          self.incoming_pipe_sender.as_deref(),
          Some(PipeMessageSender::DirectAddressed { .. })
        );
        self
          .core_pipe_manager
          .attach(rx_from_core, core_pipe_read_id_for_incoming_routing);
        tracing::info!(
          sca_handle = self.handle,
          core_read_id = core_pipe_read_id_for_incoming_routing,
          "Core pipes attached via ScaInitializePipes."
        );

        if self.current_phase == ConnectionPhaseX::WaitingForPipes {
          self.check_and_transition_to_operational().await;
        } else if self.current_phase == ConnectionPhaseX::HandshakeInProgress
          && self.zmtp_engine.phase == ZmtpPhase::Data
        {
          self.current_phase = ConnectionPhaseX::WaitingForPipes;
          self.check_and_transition_to_operational().await;
        }
      }
      Command::Stop => {
        tracing::info!(sca_handle = self.handle, "Received Stop command.");
        self.transition_to_shutdown_stream(None).await;
      }
      unexpected_cmd => {
        tracing::warn!(
          sca_handle = self.handle,
          "Received unexpected Command variant: {:?}",
          unexpected_cmd.variant_name()
        );
      }
    }
  }

  async fn process_system_event(&mut self, event: SystemEvent) {
    match event {
      SystemEvent::ContextTerminating => {
        tracing::info!(sca_handle = self.handle, "Received ContextTerminating.");
        self.transition_to_shutdown_stream(None).await;
      }
      SystemEvent::SocketClosing { socket_id } => {
        if socket_id == self.parent_socket_id {
          self.transition_to_shutdown_stream(None).await;
        }
      }
      _ => {}
    }
  }

  /// Process a completed `EngineOutput` during the ZMTP handshake phase.
  ///
  /// Writes any `NetAction::Send` bytes synchronously (within the handshake timeout),
  /// applies cork state changes, and handles `HandshakeComplete` / `PeerError`.
  async fn apply_engine_output_handshake(&mut self, output: EngineOutput) {
    let hs_timeout = self
      .zmtp_engine
      .config()
      .handshake_timeout
      .unwrap_or(Duration::from_secs(15));

    for action in output.net_actions {
      match action {
        NetAction::Send { data, .. } => {
          let wh = match self.write_half.as_mut() {
            Some(w) => w,
            None => {
              self
                .set_fatal_error(ZmqError::Internal(
                  "write_half unavailable during handshake send".into(),
                ))
                .await;
              return;
            }
          };
          match tokio::time::timeout(hs_timeout, wh.write_all(&data)).await {
            Ok(Ok(())) => {}
            Ok(Err(e)) => {
              self
                .set_fatal_error(ZmqError::from_io_endpoint(e, "hs write"))
                .await;
              return;
            }
            Err(_) => {
              self.set_fatal_error(ZmqError::Timeout).await;
              return;
            }
          }
          // Flush so the peer receives it immediately.
          let _ = tokio::time::timeout(hs_timeout, self.write_half.as_mut().unwrap().flush()).await;
        }
        NetAction::SetCork(enable) =>
        {
          #[cfg(target_os = "linux")]
          if let Some(ci) = self.cork_info.as_mut() {
            ci.apply_cork_state(enable, self.handle).await;
          }
        }
        NetAction::ScheduleClose(_) => {
          self.transition_to_shutdown_stream(None).await;
          return;
        }
      }
    }

    for action in output.app_actions {
      match action {
        AppAction::HandshakeComplete {
          peer_identity,
          peer_socket_type,
        } => {
          self.pending_peer_identity_from_handshake = peer_identity;
          self.pending_peer_socket_type = peer_socket_type;
          self.current_phase = ConnectionPhaseX::WaitingForPipes;
          self.handshake_deadline = None;
          self.check_and_transition_to_operational().await;
        }
        AppAction::PeerError(e) => {
          self.set_fatal_error(e).await;
          return;
        }
        AppAction::DeliverMessage(batch) => {
          // A data frame arrived bundled with the peer's final handshake bytes
          // (e.g. a SUB's SUBSCRIBE sent the instant it went operational). The
          // operational ingress isn't running yet, so stash it for replay rather
          // than dropping it.
          self.pending_handshake_ingress.push(batch);
        }
      }
    }
  }

  async fn check_and_transition_to_operational(&mut self) {
    if self.current_phase == ConnectionPhaseX::WaitingForPipes
      && self.zmtp_engine.phase == ZmtpPhase::Data
      && self.core_pipe_manager.is_attached()
    {
      tracing::info!(
        sca_handle = self.handle,
        "Transitioning to Operational phase."
      );
      self.current_phase = ConnectionPhaseX::Operational;

      let final_peer_identity = self.pending_peer_identity_from_handshake.take();
      let peer_socket_type = self.pending_peer_socket_type.take();

      if let Some(routing_id) = self
        .core_pipe_manager
        .state
        .core_pipe_read_id_for_incoming_routing
      {
        let event = SystemEvent::PeerIdentityEstablished {
          parent_core_id: self.parent_socket_id,
          connection_identifier: routing_id,
          peer_identity: final_peer_identity,
          peer_socket_type,
        };
        if self
          .actor_config
          .context
          .event_bus()
          .publish(event)
          .is_err()
        {
          tracing::warn!(
            sca_handle = self.handle,
            "Failed to publish PeerIdentityEstablished event."
          );
        } else {
          tracing::debug!(
            sca_handle = self.handle,
            "Published PeerIdentityEstablished event."
          );
        }
      } else {
        tracing::error!(
          sca_handle = self.handle,
          "CRITICAL: Cannot publish PeerIdentityEstablished: core_pipe_read_id not set after pipes attached."
        );
        self
          .set_fatal_error(ZmqError::Internal(
            "Pipe routing ID missing post-attach".into(),
          ))
          .await;
        return;
      }

      if let Some(monitor) = &self.actor_config.monitor_tx {
        let _ = monitor.try_send(crate::socket::SocketEvent::HandshakeSucceeded {
          endpoint: self.actor_config.connected_endpoint_uri.clone(),
        });
      }
    }
  }

  /// Fills `outgoing_batch` for one egress round, keeping the HWM/budget math
  /// in a single place: drains the `core_carryover` backlog first (a no-op when
  /// empty — the fresh-recv arm is gated on that), then tops up from the core
  /// pipe, all within the per-round `sndbatch_count`/`sndbatch_bytes` budgets
  /// and the remaining HWM budget (`sndhwm - pending_messages`, floored at 1 so
  /// a full buffer still moves one message). Messages that overflow the byte
  /// budget are pushed back to `core_carryover` for the next round.
  ///
  /// `outgoing_batch` may arrive pre-seeded (the fresh-recv arm pushes the
  /// just-awaited message before calling); seeded bytes count against the
  /// budgets.
  fn fill_outgoing_batch(
    &mut self,
    outgoing_batch: &mut Vec<FrameBatch>,
    core_carryover: &mut std::collections::VecDeque<FrameBatch>,
    pending_messages: usize,
    sndhwm: usize,
  ) {
    let wire_size = |msgs: &FrameBatch| msgs.iter().map(|m| m.size() + 9).sum::<usize>();
    let hwm_budget = sndhwm.saturating_sub(pending_messages).max(1);
    let max_count = self.zmtp_engine.config().sndbatch_count.min(hwm_budget);
    // Pre-calculated on-demand physical limit for overfill checks.
    let max_bytes = self.zmtp_engine.config().sndbatch_bytes_physical;
    let logical_max_bytes = self.zmtp_engine.config().sndbatch_bytes;

    let mut total_bytes: usize = outgoing_batch.iter().map(wire_size).sum();

    // Phase 1: drain as much as possible from the carryover backlog.
    while !core_carryover.is_empty() && outgoing_batch.len() < max_count {
      let next_msg = core_carryover.pop_front().unwrap();
      let size = wire_size(&next_msg);
      if total_bytes + size > max_bytes && !outgoing_batch.is_empty() {
        core_carryover.push_front(next_msg);
        break;
      }
      total_bytes += size;
      outgoing_batch.push(next_msg);
    }

    // Phase 2: if there is still room, top up from the core pipe.
    let start_len = outgoing_batch.len();
    if start_len < max_count && total_bytes < logical_max_bytes {
      // Dynamically calculate actual remaining slots based on the average size
      // of the messages batched so far.
      let avg_size = if start_len > 0 {
        total_bytes / start_len
      } else {
        32768
      };
      let remaining_bytes = max_bytes.saturating_sub(total_bytes);
      let needed_by_bytes = if avg_size > 0 {
        remaining_bytes / avg_size
      } else {
        0
      };

      let needed = (max_count - start_len).min(needed_by_bytes);

      if needed > 0 {
        let drained = self
          .core_pipe_manager
          .try_recv_batch_from_core(outgoing_batch, needed);

        if drained > 0 {
          let mut i = start_len;
          while i < outgoing_batch.len() {
            let size = wire_size(&outgoing_batch[i]);
            if total_bytes + size > max_bytes && i > 0 {
              // We exceeded the byte limit!
              // Move the overflow into carryover for the next cycle and stop.
              core_carryover.extend(outgoing_batch.drain(i..));
              break;
            }
            total_bytes += size;
            i += 1;
          }
        }
      }
    }
  }

  /// Frames `outgoing_batch` into `egress_buffer` for the write arm (with
  /// congestion-event bookkeeping), then applies the egress throttle.
  async fn stage_outgoing_batch(
    &mut self,
    outgoing_batch: &[FrameBatch],
    egress_buffer: &mut EgressBuffer,
    sndhwm: usize,
    adaptive_throttle: &AdaptiveThrottle,
  ) {
    if !outgoing_batch.is_empty() {
      counter!(global, global_drained_msgs, add, outgoing_batch.len() as u64);
    }

    let throttle_guard = adaptive_throttle.begin_work_bulk(
      crate::throttle::Direction::Egress,
      outgoing_batch.len() as u32,
    );

    match self.zmtp_engine.frame_batch(outgoing_batch) {
      Ok(bytes) => {
        egress_buffer.push(bytes, outgoing_batch.len());
        if !self.is_currently_congested && egress_buffer.pending_messages() >= sndhwm {
          self.is_currently_congested = true;
          if let Some(ref tx) = self.actor_config.monitor_tx {
            let ep = clean_endpoint_uri(&self.actor_config.logical_target_endpoint_uri).to_owned();
            let _ = tx.try_send(SocketEvent::ConnectionCongested { endpoint: ep });
          }
        }
      }
      Err(e) => self.set_fatal_error(e).await,
    }

    if throttle_guard.should_throttle() {
      yield_now().await;
    }
  }

  async fn set_fatal_error(&mut self, error: ZmqError) {
    tracing::error!(sca_handle = self.handle, uri = %self.actor_config.connected_endpoint_uri, err = %error, "SCA fatal error.");

    if self.error_for_drop_guard.is_none() {
      self.error_for_drop_guard = Some(error);
    }

    // Uncork before close.
    #[cfg(target_os = "linux")]
    if let Some(ci) = self.cork_info.as_mut() {
      if ci.is_corked() {
        ci.apply_cork_state(false, self.handle).await;
      }
    }

    if let Some(mut wh) = self.write_half.take() {
      match wh.shutdown().await {
        Ok(()) => {
          tracing::debug!(sca_handle = self.handle, "Write half shutdown successful.");
        }
        Err(e) => {
          tracing::warn!(sca_handle = self.handle, error = %e, "Error during write half shutdown.");
        }
      }
    }
    self.read_half = None;
    self.current_phase = ConnectionPhaseX::Terminating;
  }

  async fn transition_to_shutdown_stream(&mut self, accompanying_error: Option<ZmqError>) {
    if self.current_phase != ConnectionPhaseX::ShuttingDownStream
      && self.current_phase != ConnectionPhaseX::Terminating
    {
      if let Some(err) = accompanying_error {
        if self.error_for_drop_guard.is_none() {
          self.error_for_drop_guard = Some(err);
        }
        tracing::info!(
          sca_handle = self.handle,
          "Transitioning directly to Terminating phase due to error."
        );
        self.current_phase = ConnectionPhaseX::Terminating;
      } else if self.current_phase == ConnectionPhaseX::Operational {
        // Clean shutdown while operational: honor LINGER. Queued egress (core
        // pipe, carryover, egress buffer) must reach TCP before the FIN, so we
        // stay Operational in "draining" mode; the loop transitions to
        // ShuttingDownStream once outbound queues are empty or the linger
        // deadline expires. LINGER=0 keeps the historical drop-immediately path.
        let linger = self.zmtp_engine.config().linger;
        if linger == Some(Duration::ZERO) {
          tracing::info!(
            sca_handle = self.handle,
            "Transitioning to ShuttingDownStream phase (LINGER=0, discarding pending egress)."
          );
          self.current_phase = ConnectionPhaseX::ShuttingDownStream;
        } else if !self.graceful_drain_requested {
          self.graceful_drain_requested = true;
          self.graceful_drain_deadline = linger.map(|d| TokioInstant::now() + d);
          tracing::info!(
            sca_handle = self.handle,
            linger = ?linger,
            core_pipe_queued = self.core_pipe_manager.len(),
            "SCA clean stop — draining outbound queues before FIN (LINGER-bounded)."
          );
        }
      } else {
        tracing::info!(
          sca_handle = self.handle,
          "Transitioning to ShuttingDownStream phase."
        );
        self.current_phase = ConnectionPhaseX::ShuttingDownStream;
      }
    }
  }

  async fn perform_graceful_shutdown(&mut self) {
    tracing::info!(
      sca_handle = self.handle,
      "SCA performing graceful shutdown of stream."
    );

    let close_out = self.zmtp_engine.close();
    for action in close_out.net_actions {
      match action {
        NetAction::SetCork(false) =>
        {
          #[cfg(target_os = "linux")]
          if let Some(ci) = self.cork_info.as_mut() {
            if ci.is_corked() {
              ci.apply_cork_state(false, self.handle).await;
            }
          }
        }
        NetAction::ScheduleClose(Some(delay)) => {
          // Allow in-flight TCP segments to flush before tearing down the socket.
          tokio::time::sleep(delay).await;
        }
        _ => {}
      }
    }

    if let Some(mut wh) = self.write_half.take() {
      match wh.shutdown().await {
        Ok(()) => {
          tracing::debug!(sca_handle = self.handle, "Write half shutdown successful.");
        }
        Err(e) => {
          tracing::warn!(sca_handle = self.handle, error = %e, "Error during write half shutdown.");
          if self.error_for_drop_guard.is_none() {
            self.error_for_drop_guard = Some(ZmqError::from_io_endpoint(e, "graceful shutdown"));
          }
        }
      }
    }
    self.read_half = None;
    self.perform_final_cleanup_actions().await;
  }

  async fn perform_final_cleanup_actions(&mut self) {
    // If halves were taken back from the operational loop but not yet shut down, close them.
    if self.write_half.is_some() || self.read_half.is_some() {
      // No need to call engine.close() again if already called from graceful shutdown.
      if let Some(mut wh) = self.write_half.take() {
        let _ = wh.shutdown().await;
      }
      self.read_half = None;
    }
    self.core_pipe_manager.detach_and_clear_pipes();
    self.current_phase = ConnectionPhaseX::Terminating;
    tracing::debug!(
      sca_handle = self.handle,
      "SCA final cleanup actions complete."
    );
  }
}