rzmq 0.5.18

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

use crate::error::ZmqError;
use crate::message::Msg;
use crate::runtime::{ActorDropGuard, ActorType, Command, SystemEvent};
use crate::sessionx::regulator::SessionRegulator;
use crate::socket::ISocket;
use crate::socket::options::ZmtpEngineConfig;
use crate::throttle::AdaptiveThrottle;
use crate::transport::{ZmtpStdStream, ZmtpWriteHalf};
use crate::{Blob, MailboxReceiver};

use futures::FutureExt;
use std::fmt::Debug;
#[cfg(target_os = "linux")]
use std::os::fd::AsRawFd; // For AsRawFd bound if S needs it for cork_info init
use std::sync::Arc;
use std::time::Duration;
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::pipe_manager::CorePipeManagerX;
use super::protocol_handler::ZmtpProtocolHandlerX;
use super::states::ActorConfigX;
use super::types::{ConnectionPhaseX, ZmtpHandshakeProgressX};

// SessionConnectionActorX might not need to be Debug itself if not printed directly.
// If it is, S would need to be Debug. Let's assume for now it doesn't need direct Debug.
pub(crate) struct SessionConnectionActorX<S: ZmtpStdStream> {
  handle: usize,
  current_phase: ConnectionPhaseX,
  parent_socket_id: usize,
  actor_config: ActorConfigX,

  zmtp_handler: ZmtpProtocolHandlerX<S>,
  core_pipe_manager: CorePipeManagerX,

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

  pending_peer_identity_from_handshake: Option<Blob>,
  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>,
}

// Add AsRawFd bound here if ZmtpProtocolHandlerX::new needs it for cork setup.
// This depends on how try_create_cork_info and ZPHX::new are finalized for S.
// If ZPHX::new takes Option<RawFd>, then SCA doesn't need this bound directly for that.
// However, if ZPHX::new takes &S and calls as_raw_fd(), then SCA::new needs S: AsRawFd.
// Let's assume for now that ZPHX::new is passed what it needs regarding FD for cork.
impl<S> SessionConnectionActorX<S>
where
  S: ZmtpStdStream + Debug + Unpin + Send + 'static, // Keep Send + 'static for tokio::spawn
{
  pub(crate) fn create_and_spawn(
    handle: usize,
    parent_socket_id: usize,
    stream: S, // Consumes the stream
    actor_config: ActorConfigX,
    engine_config: Arc<ZmtpEngineConfig>,
    command_mailbox_receiver: MailboxReceiver,
    socket_logic: Arc<dyn ISocket>,
    connection_permit: Option<OwnedSemaphorePermit>,
  ) -> JoinHandle<()> {
    // Returns JoinHandle for the SCA task

    let zmtp_handler = ZmtpProtocolHandlerX::new(
      stream,
      engine_config.clone(), // ZmtpProtocolHandlerX clones Arc
      actor_config.is_server_role,
      handle, // Pass actor handle for logging within ZmtpProtocolHandlerX
    );

    let mut ping_check_timer = None;
    if let Some(ivl) = engine_config.heartbeat_ivl {
      if !ivl.is_zero() {
        // Start timer such that the first tick is after `ivl` from now.
        let mut timer = tokio::time::interval_at(TokioInstant::now() + ivl, ivl);
        timer.set_missed_tick_behavior(MissedTickBehavior::Delay);
        ping_check_timer = Some(timer);
      }
    }

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

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

    // We enforce a minimum 1s lifespan if the configured receive timeout is short,
    // to prevent rapid crash-loops from flooding logs.
    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 {
      // Even if rcvtimeo is long, a crash-on-connect should still be throttled.
      std::time::Duration::from_millis(1000)
    };

    let actor = Self {
      handle,
      current_phase: ConnectionPhaseX::HandshakeInProgress, // Start handshake immediately
      parent_socket_id,
      actor_config,
      zmtp_handler,
      core_pipe_manager: CorePipeManagerX::new(),
      command_mailbox_receiver,
      system_event_receiver,
      pending_peer_identity_from_handshake: None,
      error_for_drop_guard: None,
      ping_check_timer,
      handshake_deadline,
      socket_logic,
      session_regulator: SessionRegulator::new(regulator_min_lifespan),
      _connection_permit: connection_permit,
    };

    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, // Consider ActorType::SessionConnectionActorX
      Some(self.actor_config.connected_endpoint_uri.clone()),
      Some(self.parent_socket_id),
    );

    let adaptive_throttle = {
      let mut config = self.zmtp_handler.config.throttle_config.clone();
      // Priority is role-dependent and cannot be set via socket options.
      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)
    };

    // Reusable batch buffer — allocated once, cleared on each wakeup
    let mut outgoing_batch: Vec<Vec<Msg>> =
      Vec::with_capacity(self.zmtp_handler.config.sndbatch_count);

    let mut egress_buffer = EgressBuffer::new();
    const EGRESS_BACKPRESSURE_BYTES: usize = 256 * 1024;

    // ── HANDSHAKE LOOP ────────────────────────────────────────────────────────
    // zmtp_handler owns both halves here; advance_handshake() drives all I/O.
    // Exits when phase leaves HandshakeInProgress/WaitingForPipes, or on shutdown.
    'handshake: while matches!(
      self.current_phase,
      ConnectionPhaseX::HandshakeInProgress | ConnectionPhaseX::WaitingForPipes
    ) {
      if self.current_phase == ConnectionPhaseX::ShuttingDownStream {
        self.perform_graceful_shutdown().await;
        break 'handshake;
      }

      let mut overall_handshake_timeout_future = futures::future::pending().left_future();
      if let Some(deadline) = self.handshake_deadline {
        overall_handshake_timeout_future = tokio::time::sleep_until(deadline).right_future();
      }

      tokio::select! {
        biased;

        maybe_cmd = self.command_mailbox_receiver.recv() => {
          match maybe_cmd {
            Ok(command) => self.process_command(command).await,
            Err(_) => {
              tracing::info!(sca_handle = self.handle, "SCA Command mailbox closed.");
              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;
            }
          }
        }

        _ = overall_handshake_timeout_future, if self.handshake_deadline.is_some() => {
          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;
        }

        handshake_res = self.zmtp_handler.advance_handshake(),
            if self.current_phase == ConnectionPhaseX::HandshakeInProgress => {
          match handshake_res {
            Ok(p) => self.handle_handshake_progression(p).await,
            Err(e) => self.set_fatal_error(e).await,
          }
        }
      }
    }

    // ── OPERATIONAL LOOP ──────────────────────────────────────────────────────
    // Split halves into local variables so ingress and egress arms are independent.
    if self.current_phase == ConnectionPhaseX::Operational {
      let mut read_half = self.zmtp_handler.read_half.take()
        .expect("read_half must be present at operational start");
      let mut write_half = self.zmtp_handler.write_half.take()
        .expect("write_half must be present at operational start");

      let use_owned_write = write_half.supports_owned_write();
      let mut pending_vectored: std::collections::VecDeque<Vec<bytes::Bytes>> =
        std::collections::VecDeque::new();

      'operational: while self.current_phase == ConnectionPhaseX::Operational
        || (self.current_phase == ConnectionPhaseX::ShuttingDownStream
            && (!egress_buffer.is_empty() || !pending_vectored.is_empty()))
      {
        let mut pong_timeout_future = futures::future::pending().left_future();
        if self.zmtp_handler.heartbeat_state.waiting_for_pong {
          if let Some(deadline_std) = self.zmtp_handler.heartbeat_state.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() => {
            match maybe_cmd {
              Ok(command) => self.process_command(command).await,
              Err(_) => {
                tracing::info!(sca_handle = self.handle, "SCA Command mailbox closed.");
                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() => {
            match self.zmtp_handler.maybe_build_ping() {
              Ok(Some(bytes)) => egress_buffer.push_priority(bytes),
              Ok(None) => {}
              Err(e) => self.set_fatal_error(e).await,
            }
          }

          _ = pong_timeout_future, if self.zmtp_handler.heartbeat_state.waiting_for_pong => {
            if self.zmtp_handler.has_pong_timed_out() {
              self.set_fatal_error(ZmqError::Timeout).await;
            }
          }

          // Ingress: concurrent with egress — independent local borrow
          ingress_res = self.zmtp_handler.read_and_parse_data_frames_batch(&mut read_half) => {
            match ingress_res {
              Ok(msgs) if !msgs.is_empty() => {
                let pipe_read_id = self
                  .core_pipe_manager
                  .state
                  .core_pipe_read_id_for_incoming_routing
                  .expect("pipe_id required for incoming routing");

                let mut data_msgs = Vec::with_capacity(msgs.len());
                for msg in msgs {
                  if msg.is_command() {
                    match self.zmtp_handler.process_incoming_data_command_frame(&msg) {
                      Ok(Some(pong)) => {
                        match self.zmtp_handler.frame_outgoing_msgs(vec![pong]) {
                          Ok(bytes) => egress_buffer.push_priority(bytes),
                          Err(e) => { self.set_fatal_error(e).await; }
                        }
                      }
                      Ok(None) => {}
                      Err(e) => { self.set_fatal_error(e).await; }
                    }
                  } else {
                    data_msgs.push(msg);
                  }
                }

                if !data_msgs.is_empty() {
                  let weight = data_msgs.len() as u32;
                  let throttle_guard = adaptive_throttle
                    .begin_work_bulk(crate::throttle::Direction::Ingress, weight);

                  let cmd = crate::runtime::Command::PipeMessageBatchReceived {
                    pipe_id: pipe_read_id,
                    msgs: data_msgs,
                  };
                  if let Err(e) = self.socket_logic.handle_pipe_event(pipe_read_id, cmd).await {
                    self.set_fatal_error(e).await;
                  }

                  if throttle_guard.should_throttle() {
                    yield_now().await;
                  }
                }
              }
              Ok(_) => {}
              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,
            }
          }

          // Unified write arm: priority egress (EgressBuffer pings/pongs) drains before
          // zero-copy data. Only one arm borrows write_half so the borrow checker is happy.
          // Some(n) = AsyncWrite bytes advanced; None = write_owned completed.
          write_res = async {
            use tokio::io::AsyncWriteExt;
            if egress_buffer.current_slice().is_some() {
              let n = write_half.write(egress_buffer.current_slice().unwrap()).await?;
              Ok::<Option<usize>, std::io::Error>(Some(n))
            } else {
              let bufs = pending_vectored.front().cloned().unwrap_or_default();
              write_half.write_owned(bufs).await?;
              Ok(None)
            }
          }, if egress_buffer.current_slice().is_some()
              || (use_owned_write && pending_vectored.front().is_some()) => {
            match write_res {
              Ok(Some(n)) => {
                egress_buffer.advance(n);
                self.zmtp_handler.heartbeat_state.record_activity();
              }
              Ok(None) => {
                pending_vectored.pop_front();
                self.zmtp_handler.heartbeat_state.record_activity();
              }
              Err(e) => {
                self.set_fatal_error(ZmqError::from_io_endpoint(e, "egress write")).await;
              }
            }
          }

          // Outgoing from SocketCore - Only accept messages while in Operational phase
          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()
                && if use_owned_write {
                  pending_vectored.is_empty()
                } else {
                  egress_buffer.total_pending_bytes() < EGRESS_BACKPRESSURE_BYTES
                } => {
            match maybe_msgs_from_core {
              Ok(first_msgs) => {
                outgoing_batch.clear();
                // Count wire bytes (payload + 9 bytes ZMTP long-frame overhead per message)
                // so the framed batch never exceeds sndbatch_bytes and busts the ZC slot.
                let wire_size = |msgs: &Vec<Msg>| msgs.iter().map(|m| m.size() + 9).sum::<usize>();
                let mut total_bytes = wire_size(&first_msgs);
                let mut total_msgs = 1usize;
                outgoing_batch.push(first_msgs);

                let max_count = self.zmtp_handler.config.sndbatch_count;
                let max_bytes = self.zmtp_handler.config.sndbatch_bytes;

                while total_msgs < max_count && total_bytes < max_bytes {
                  match self.core_pipe_manager.try_recv_from_core() {
                    Ok(next_msgs) => {
                      total_bytes += wire_size(&next_msgs);
                      total_msgs += 1;
                      outgoing_batch.push(next_msgs);
                    }
                    Err(_) => break,
                  }
                }

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

                if use_owned_write {
                  match self.zmtp_handler.frame_outgoing_batch_vectored(&outgoing_batch) {
                    Ok(bufs) => pending_vectored.push_back(bufs),
                    Err(e) => self.set_fatal_error(e).await,
                  }
                } else {
                  match self.zmtp_handler.frame_outgoing_batch(&outgoing_batch) {
                    Ok(bytes) => egress_buffer.push(bytes),
                    Err(e) => self.set_fatal_error(e).await,
                  }
                }

                if throttle_guard.should_throttle() {
                  yield_now().await;
                }
              }
              Err(_) => {
                tracing::info!(sca_handle = self.handle, "Pipe from SocketCore closed.");
                self.transition_to_shutdown_stream(Some(ZmqError::ConnectionClosed)).await;
              }
            }
          }

        }
      }

      // Restore halves so initiate_stream_shutdown can shut down the write half.
      self.zmtp_handler.read_half = Some(read_half);
      self.zmtp_handler.write_half = Some(write_half);
    }

    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
    );

    // Post-loop cleanup (already initiated if error/stop occurred, this is a final catch-all)
    if self.current_phase != ConnectionPhaseX::Terminating {
      // If loop exited for other reasons
      self.perform_final_cleanup_actions().await;
    }

    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();
    }

    // This publishes the ActorStopping event (which triggers the Disconnected monitor event)
    // IMMEDIATELY, before we potentially sleep for the churn penalty.
    drop(actor_drop_guard);

    // Now, enforce the minimum lifespan penalty if there was an error.
    // The rest of the system is already notified that we are dead.
    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 for select! Arms & State Transitions ---

  async fn process_command(&mut self, command: Command) {
    tracing::debug!(
      sca_handle = self.handle,
      "Processing ScaCommand: {:?}",
      command
    );
    match command {
      Command::ScaInitializePipes {
        sca_handle_id, // Used to verify command is for this actor instance
        rx_from_core,
        core_pipe_read_id_for_incoming_routing,
      } => {
        if self.handle != sca_handle_id {
          tracing::warn!(
            sca_handle = self.handle,
            expected_id = sca_handle_id,
            "Received ScaInitializePipes for wrong actor. Ignoring."
          );
          // Important: Drop rx_from_core to prevent resource leak if it's not used
          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); // Still drop the provided channels
          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
          .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_handler.is_handshake_complete()
        {
          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;
      }
      _ => {}
    }
  }

  async fn handle_handshake_progression(&mut self, progress: ZmtpHandshakeProgressX) {
    if self.current_phase != ConnectionPhaseX::HandshakeInProgress {
      return;
    }

    match progress {
      ZmtpHandshakeProgressX::InProgress => {}
      ZmtpHandshakeProgressX::IdentityReady(blob) => {
        self.pending_peer_identity_from_handshake = Some(blob);
        if !self.zmtp_handler.is_handshake_complete() {
          return;
        }
      }
      ZmtpHandshakeProgressX::HandshakeComplete => {}
      ZmtpHandshakeProgressX::ProtocolError(s) => {
        self.set_fatal_error(ZmqError::ProtocolViolation(s)).await;
        return;
      }
      ZmtpHandshakeProgressX::FatalError(e) => {
        self.set_fatal_error(e).await;
        return;
      }
    }

    if self.zmtp_handler.is_handshake_complete() {
      tracing::info!(
        sca_handle = self.handle,
        "Handshake complete via ZmtpProtocolHandlerX."
      );
      self.current_phase = ConnectionPhaseX::WaitingForPipes;
      self.handshake_deadline = None;
      self.check_and_transition_to_operational().await;
    }
  }

  async fn check_and_transition_to_operational(&mut self) {
    if self.current_phase == ConnectionPhaseX::WaitingForPipes &&
       self.zmtp_handler.is_handshake_complete() && // Should always be true if in WaitingForPipes
       self.core_pipe_manager.is_attached()
    {
      tracing::info!(
        sca_handle = self.handle,
        "Transitioning to Operational phase."
      );
      self.current_phase = ConnectionPhaseX::Operational;

      // Deferred cork: handshake is complete, now safe to enable TCP_CORK for
      // streaming patterns. Applied here rather than at socket creation so that
      // small handshake frames are never held by the 200ms kernel cork timer.
      let socket_type = self.zmtp_handler.config.socket_type_name.as_str();
      if self.zmtp_handler.config.use_cork
        && matches!(socket_type, "PUSH" | "PULL" | "PUB" | "SUB")
      {
        if let Some(ref wh) = self.zmtp_handler.write_half {
          wh.set_cork(true);
        }
        #[cfg(target_os = "linux")]
        if let Some(ref mut ci) = self.zmtp_handler.cork_info {
          ci.apply_cork_state(true, self.handle).await;
        }
      }

      // Determine final peer identity: one from handshake (security or READY) takes precedence.
      let final_peer_identity = self.pending_peer_identity_from_handshake.take();
      let peer_socket_type = self.zmtp_handler.handshake_state.peer_socket_type.clone();

      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."
        );
        // This is a logic error if pipes are attached but no routing ID.
        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(),
        });
      }
    }
  }

  async fn handle_incoming_from_network(&mut self, msg: Msg) {
    // Get the pipe_id needed for ISocket::handle_pipe_event
    let pipe_read_id = self
      .core_pipe_manager
      .state
      .core_pipe_read_id_for_incoming_routing
      .expect("Cannot handle incoming message without core_pipe_read_id");

    if msg.is_command() {
      // Process PING/PONG
      match self.zmtp_handler.process_incoming_data_command_frame(&msg) {
        Ok(Some(pong_reply)) => {
          if let Err(e) = self.zmtp_handler.write_data_msg(pong_reply, true).await {
            self.set_fatal_error(e).await;
          }
        }
        Ok(None) => { /* PONG received and handled */ }
        Err(e) => self.set_fatal_error(e).await,
      }
    } else {
      // Data message. Send DIRECTLY to ISocket pattern logic.
      let command_for_isocket = Command::PipeMessageReceived {
        pipe_id: pipe_read_id,
        msg,
      };
      if let Err(e) = self
        .socket_logic
        .handle_pipe_event(pipe_read_id, command_for_isocket)
        .await
      {
        tracing::error!(sca_handle = self.handle, pipe_id = pipe_read_id, error = %e, "Fatal error from ISocket::handle_pipe_event. Shutting down connection.");
        self.set_fatal_error(e).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);
    }

    // Explicitly shutdown and drop the stream NOW to free the OS File Descriptor.
    let _ = self.zmtp_handler.initiate_stream_shutdown().await;

    // Transition directly to Terminating to exit the loop immediately since the stream is dead.
    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 {
        tracing::info!(
          sca_handle = self.handle,
          "Transitioning to ShuttingDownStream phase."
        );
        self.current_phase = ConnectionPhaseX::ShuttingDownStream;
      }
    }
  }

  async fn perform_graceful_shutdown(&mut self) {
    // This method is called when self.current_phase is ShuttingDownStream
    tracing::info!(
      sca_handle = self.handle,
      "SCA performing graceful shutdown of stream."
    );
    if let Err(e) = self.zmtp_handler.initiate_stream_shutdown().await {
      tracing::warn!(
        sca_handle = self.handle,
        "Error during stream shutdown: {}.",
        e
      );
      if self.error_for_drop_guard.is_none() {
        self.error_for_drop_guard = Some(e);
      }
    }
    self.perform_final_cleanup_actions().await; // Includes setting phase to Terminating
  }

  async fn perform_final_cleanup_actions(&mut self) {
    if self.zmtp_handler.read_half.is_some() || self.zmtp_handler.write_half.is_some() {
      let _ = self.zmtp_handler.initiate_stream_shutdown().await;
    }
    self.core_pipe_manager.detach_and_clear_pipes();
    self.current_phase = ConnectionPhaseX::Terminating;
    tracing::debug!(
      sca_handle = self.handle,
      "SCA final cleanup actions complete."
    );
  }
}