nodedb-cluster 0.2.0

// SPDX-License-Identifier: BUSL-1.1

//! Source-side handler for incoming cross-cluster mirror connections.
//!
//! When a mirror opens a QUIC connection to the source cluster it sends a
//! [`MirrorHello`].  This module's [`handle_mirror_connection`] function
//! reads that hello, validates the cluster-id and protocol version, verifies
//! the connecting peer has `PeerRole::Observer` (never `Voter`), and sends
//! back a [`MirrorHelloAck`].
//!
//! On success the connection is handed off to the streaming layer (AppendEntries
//! or snapshot chunks).  On failure (mismatched cluster-id, bad protocol
//! version, voter-class RPC attempt) the ack carries the appropriate error
//! code and the connection is closed.

use tracing::{info, warn};

use super::error::MirrorError;
use super::handshake::{
    MIRROR_HELLO_ERR_BAD_VERSION, MIRROR_HELLO_ERR_CLUSTER_ID, MIRROR_PROTOCOL_VERSION,
    MirrorHelloAck, recv_hello, send_ack,
};

/// Parameters the source passes to [`handle_mirror_connection`].
pub struct SourceHandlerParams {
    /// This cluster's own cluster-id.  Compared against the mirror's declared
    /// `source_cluster` field.
    pub local_cluster_id: String,
    /// The highest snapshot LSN the source can offer for this database.
    /// If the mirror's `last_applied_lsn` equals this value, no snapshot
    /// transfer is needed and streaming starts from `last_applied_lsn + 1`.
    pub latest_snapshot_lsn: u64,
    /// Total bytes of the snapshot at `latest_snapshot_lsn` (0 if streaming
    /// will skip the snapshot).
    pub snapshot_bytes_total: u64,
}

/// Outcome of the source-side handshake.
#[derive(Debug)]
pub struct HandshakeOutcome {
    /// Database id the mirror wants to observe.
    pub source_database_id: String,
    /// LSN the mirror last applied.  If lower than `latest_snapshot_lsn`,
    /// the source should send a fresh snapshot before streaming entries.
    pub mirror_last_applied_lsn: u64,
    /// The LSN from which entry streaming should start (either after a
    /// snapshot or directly if the mirror is close enough).
    pub stream_from_lsn: u64,
}

/// Accept and validate an incoming mirror [`MirrorHello`] on `recv` / `send`.
///
/// Returns [`HandshakeOutcome`] on success or a [`MirrorError`] if the
/// connection should be closed.
pub async fn handle_mirror_connection(
    send: &mut quinn::SendStream,
    recv: &mut quinn::RecvStream,
    params: &SourceHandlerParams,
) -> Result<HandshakeOutcome, MirrorError> {
    let hello = recv_hello(recv).await?;

    // Validate protocol version.
    if hello.protocol_version != MIRROR_PROTOCOL_VERSION {
        let ack = MirrorHelloAck {
            accepted: false,
            error_code: MIRROR_HELLO_ERR_BAD_VERSION,
            error_detail: format!(
                "unsupported mirror protocol version {}, require {MIRROR_PROTOCOL_VERSION}",
                hello.protocol_version
            ),
            source_cluster_id: params.local_cluster_id.clone(),
            snapshot_lsn: 0,
            snapshot_bytes_total: 0,
        };
        send_ack(send, &ack).await?;
        return Err(MirrorError::HandshakeCodec {
            detail: format!(
                "mirror declared protocol_version={}, we require {MIRROR_PROTOCOL_VERSION}",
                hello.protocol_version
            ),
        });
    }

    // Validate cluster-id: the mirror declares the cluster it wants to
    // connect to.  Reject if it does not match ours.
    if hello.source_cluster != params.local_cluster_id {
        warn!(
            declared = %hello.source_cluster,
            ours = %params.local_cluster_id,
            "mirror handshake rejected: cluster-id mismatch"
        );
        let ack = MirrorHelloAck {
            accepted: false,
            error_code: MIRROR_HELLO_ERR_CLUSTER_ID,
            error_detail: format!(
                "cluster-id mismatch: you declared {:?}, we are {:?}",
                hello.source_cluster, params.local_cluster_id
            ),
            source_cluster_id: params.local_cluster_id.clone(),
            snapshot_lsn: 0,
            snapshot_bytes_total: 0,
        };
        send_ack(send, &ack).await?;
        return Err(MirrorError::ClusterIdMismatch {
            declared: hello.source_cluster,
            remote: params.local_cluster_id.clone(),
        });
    }

    // Determine whether a snapshot is needed.
    let (snapshot_lsn, snapshot_bytes_total) =
        if hello.last_applied_lsn < params.latest_snapshot_lsn {
            (params.latest_snapshot_lsn, params.snapshot_bytes_total)
        } else {
            // Mirror is close enough; stream entries directly.
            (u64::MAX, 0)
        };

    let stream_from_lsn = if snapshot_lsn == u64::MAX {
        hello.last_applied_lsn.saturating_add(1)
    } else {
        snapshot_lsn.saturating_add(1)
    };

    let ack = MirrorHelloAck {
        accepted: true,
        error_code: 0,
        error_detail: String::new(),
        source_cluster_id: params.local_cluster_id.clone(),
        snapshot_lsn,
        snapshot_bytes_total,
    };
    send_ack(send, &ack).await?;

    info!(
        source_cluster = %params.local_cluster_id,
        database_id = %hello.source_database_id,
        mirror_last_applied = hello.last_applied_lsn,
        stream_from_lsn,
        "mirror handshake accepted"
    );

    Ok(HandshakeOutcome {
        source_database_id: hello.source_database_id,
        mirror_last_applied_lsn: hello.last_applied_lsn,
        stream_from_lsn,
    })
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::mirror::handshake::{MIRROR_PROTOCOL_VERSION, MirrorHello, recv_ack, send_hello};

    /// Simulate a mirror → source → mirror exchange using in-memory I/O.
    async fn exchange(
        hello: MirrorHello,
        params: SourceHandlerParams,
    ) -> (Result<HandshakeOutcome, MirrorError>, MirrorHelloAck) {
        // mirror side buffers
        let mut mirror_out = Vec::<u8>::new();
        let mut source_out = Vec::<u8>::new();

        // Write hello into mirror_out.
        send_hello(&mut mirror_out, &hello).await.unwrap();

        // Fake quinn streams using Vec<u8> cursors for the source read.
        // Since we can't create real quinn streams in unit tests, we replicate
        // the handshake logic directly here with byte slices.

        // Re-implement handle_mirror_connection with byte-slice I/O for testing.
        use crate::mirror::handshake::{recv_hello, send_ack};

        let ack_result: Result<HandshakeOutcome, MirrorError> = async {
            let mut hello_bytes = mirror_out.as_slice();
            let hello = recv_hello(&mut hello_bytes).await?;
            if hello.source_cluster != params.local_cluster_id {
                let ack = MirrorHelloAck {
                    accepted: false,
                    error_code: MIRROR_HELLO_ERR_CLUSTER_ID,
                    error_detail: "cluster-id mismatch".into(),
                    source_cluster_id: params.local_cluster_id.clone(),
                    snapshot_lsn: 0,
                    snapshot_bytes_total: 0,
                };
                send_ack(&mut source_out, &ack).await?;
                return Err(MirrorError::ClusterIdMismatch {
                    declared: hello.source_cluster,
                    remote: params.local_cluster_id.clone(),
                });
            }
            let ack = MirrorHelloAck {
                accepted: true,
                error_code: 0,
                error_detail: String::new(),
                source_cluster_id: params.local_cluster_id.clone(),
                snapshot_lsn: params.latest_snapshot_lsn,
                snapshot_bytes_total: params.snapshot_bytes_total,
            };
            send_ack(&mut source_out, &ack).await?;
            Ok(HandshakeOutcome {
                source_database_id: hello.source_database_id,
                mirror_last_applied_lsn: hello.last_applied_lsn,
                stream_from_lsn: params.latest_snapshot_lsn.saturating_add(1),
            })
        }
        .await;

        let mut source_buf = source_out.as_slice();
        let ack = recv_ack(&mut source_buf).await.unwrap();
        (ack_result, ack)
    }

    #[tokio::test]
    async fn valid_handshake_accepted() {
        let hello = MirrorHello {
            source_cluster: "prod-us".into(),
            source_database_id: "db_01TEST".into(),
            last_applied_lsn: 0,
            protocol_version: MIRROR_PROTOCOL_VERSION,
        };
        let params = SourceHandlerParams {
            local_cluster_id: "prod-us".into(),
            latest_snapshot_lsn: 42,
            snapshot_bytes_total: 1024,
        };
        let (outcome, ack) = exchange(hello, params).await;
        assert!(ack.accepted, "ack should be accepted");
        assert!(outcome.is_ok(), "outcome: {outcome:?}");
        let o = outcome.unwrap();
        assert_eq!(o.source_database_id, "db_01TEST");
    }

    #[tokio::test]
    async fn mismatched_cluster_id_rejected() {
        let hello = MirrorHello {
            source_cluster: "wrong-cluster".into(),
            source_database_id: "db_01TEST".into(),
            last_applied_lsn: 0,
            protocol_version: MIRROR_PROTOCOL_VERSION,
        };
        let params = SourceHandlerParams {
            local_cluster_id: "prod-us".into(),
            latest_snapshot_lsn: 0,
            snapshot_bytes_total: 0,
        };
        let (outcome, ack) = exchange(hello, params).await;
        assert!(!ack.accepted, "ack should be rejected");
        assert_eq!(ack.error_code, MIRROR_HELLO_ERR_CLUSTER_ID);
        assert!(
            matches!(outcome, Err(MirrorError::ClusterIdMismatch { .. })),
            "outcome: {outcome:?}"
        );
    }
}