laminar_core/shuffle/transport/grpc/
pipeline_identity.rs

1//! Bind a process's shuffle fabric to its pipeline identity, so peers compiled from another
2//! binary or catalog refuse to connect instead of exchanging frames they encode differently.
3
4use super::*;
5use crate::checkpoint::PipelineIdentity;
6
7impl ShuffleSender {
8    /// Bind both directions of this process's fabric to `identity`. A receiver refuses the
9    /// handshake of a sender bound to a different identity or to none, so the first exchange
10    /// fails before any frame crosses. Changing the identity cancels the streams and handshakes
11    /// opened under the previous one; sequence domains stay scoped to the assignment.
12    ///
13    /// # Errors
14    /// Rejects a non-canonical identity and a sender and receiver of different processes.
15    pub fn install_pipeline_identity_pair(
16        &self,
17        receiver: &ShuffleReceiver,
18        identity: &PipelineIdentity,
19    ) -> io::Result<bool> {
20        let digest = identity_digest(identity)?;
21        // Same lock order as assignment and topology installation, which open and admit streams.
22        let _sender_assignment = self.assignment.write();
23        let _receiver_assignment = receiver.assignment.write();
24        if self.local_id != receiver.local_id
25            || self.sender_incarnation != receiver.receiver_incarnation
26        {
27            return Err(io::Error::new(
28                io::ErrorKind::PermissionDenied,
29                "pipeline identity binds one process's own sender and receiver",
30            ));
31        }
32        let mut sender_identity = self.pipeline_identity.write();
33        let mut receiver_identity = receiver.pipeline_identity.write();
34        if *sender_identity == Some(digest) && *receiver_identity == Some(digest) {
35            return Ok(false);
36        }
37        *sender_identity = Some(digest);
38        *receiver_identity = Some(digest);
39        for (scope, lease) in [
40            (&self.scope_cancel, &self.process_lease),
41            (&receiver.scope_cancel, &receiver.process_lease),
42        ] {
43            if !scope.read().is_cancelled() {
44                rotate_scope_token(scope, lease, true);
45            }
46        }
47        self.pool.lock().clear();
48        self.connect_locks.lock().clear();
49        receiver.pending_handshakes.clear();
50        Ok(true)
51    }
52}
53
54fn identity_digest(identity: &PipelineIdentity) -> io::Result<[u8; 32]> {
55    if !identity.is_canonical() {
56        return Err(io::Error::new(
57            io::ErrorKind::InvalidInput,
58            "shuffle pipeline identity is not canonical",
59        ));
60    }
61    let mut digest = [0; 32];
62    for (byte, encoded) in digest
63        .iter_mut()
64        .zip(identity.sha256.as_bytes().as_chunks::<2>().0)
65    {
66        let encoded = std::str::from_utf8(encoded).map_err(io::Error::other)?;
67        *byte = u8::from_str_radix(encoded, 16).map_err(io::Error::other)?;
68    }
69    Ok(digest)
70}
laminar_core/shuffle/transport/grpc/pipeline_identity.rs

laminar_core/shuffle/transport/grpc/
pipeline_identity.rs
