liminal-sdk 0.19.0

Application-facing SDK traits for liminal messaging clients
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80

//! A quiet drain ends on an on-demand transport fence, not elapsed silence.
//! The fence owns no participant authority: a Pong is neither a delivery nor
//! marker settlement nor a receipt. Each drain sends at most one fresh probe;
//! replies left by earlier delivery returns are consumed in wire order.

use alloc::format;
use core::time::Duration;
use std::time::Instant;

use liminal::protocol::Frame;
use liminal_protocol::wire::ParticipantFrame;

use super::{Connection, FrameStream, IO_TIMEOUT, ParticipantPumpDiagnostics, participant};
use crate::SdkError;

impl<S: FrameStream> Connection<S> {
    /// Consume only replies owned by this link's pump probes. Other Pongs stay
    /// visible to the ordinary protocol-direction checks.
    pub(super) const fn consume_pump_fence(&mut self, frame: &Frame) -> bool {
        if matches!(frame, Frame::Pong { .. }) && self.pump_fences_pending > 0 {
            self.pump_fences_pending -= 1;
            true
        } else {
            false
        }
    }

    /// What this connection's quiet drains have done.
    pub(in crate::remote) const fn pump_diagnostics(&self) -> ParticipantPumpDiagnostics {
        self.pump_diagnostics
    }

    pub(in crate::remote) fn try_receive_participant(
        &mut self,
    ) -> Result<Option<ParticipantFrame>, SdkError> {
        self.pump_diagnostics.pumps_run = self.pump_diagnostics.pumps_run.saturating_add(1);
        // Already-decoded participant traffic wins without a round trip. This
        // also retires an earlier fence if a different receive left it behind.
        if let Some(frame) = self.receive_optional_within(Duration::ZERO)? {
            return participant::response_frame(frame).map(Some);
        }
        let pending =
            self.pump_fences_pending
                .checked_add(1)
                .ok_or_else(|| SdkError::Protocol {
                    description: "participant pump fence count exhausted".into(),
                })?;
        let started = Instant::now();
        self.send(&Frame::Ping { flags: 0 })?;
        self.pump_fences_pending = pending;
        let result = self.poll_within(started, IO_TIMEOUT, true);
        let restore =
            self.stream
                .set_read_deadline(IO_TIMEOUT)
                .map_err(|error| SdkError::Connection {
                    description: format!(
                        "failed to restore read timeout after pump fence: {error}"
                    ),
                });
        let frame = result?;
        restore?;
        match frame {
            Some(frame) => participant::response_frame(frame).map(Some),
            None if self.pump_fences_pending == 0 => {
                // The same socket read may also have decoded traffic following
                // the Pong. Return it rather than claiming quiet over a frame
                // already in our own buffer; do not arm another read.
                self.receive_optional_within(Duration::ZERO)?
                    .map(participant::response_frame)
                    .transpose()
            }
            None => Err(SdkError::Connection {
                description: format!(
                    "participant pump liveness fence was unanswered after {:.3}s",
                    started.elapsed().as_secs_f64()
                ),
            }),
        }
    }
}