openlogi_device/replay/
channel.rs

1use std::collections::{HashMap, VecDeque};
2use std::error::Error;
3use std::io;
4use std::sync::atomic::{AtomicBool, Ordering};
5use std::sync::{Arc, Mutex, PoisonError};
6
7use hidpp::channel::RawHidChannel;
8use openlogi_fixture::{HidCassette, ReportSupport, RequestMatch, format_hex};
9use tokio::sync::mpsc;
10
11use crate::backend::{BackendError, RawWriter};
12
13use super::barrier::{RequestKey, ResponseGates};
14use super::slots::ReceiverSlots;
15use super::{ReceiverSlot, ReceiverSlotState, ReplayError};
16
17type Responder = Arc<dyn Fn(&[u8]) -> Option<Vec<u8>> + Send + Sync>;
18
19impl RequestKey {
20    fn description(&self) -> String {
21        match self {
22            Self::Exact(bytes) => format!("exact:{}", format_hex(bytes)),
23            Self::Hidpp20(bytes) => format!("hidpp20:{}", format_hex(bytes)),
24        }
25    }
26}
27
28struct PendingExchange {
29    key: RequestKey,
30    response: Option<Vec<u8>>,
31    required: bool,
32    consumed: bool,
33}
34
35struct CassetteRuntime {
36    exchanges: Vec<PendingExchange>,
37    queues: HashMap<RequestKey, VecDeque<usize>>,
38    unmatched: Vec<ReplayMismatch>,
39    receiver_slots: ReceiverSlots,
40    slot_state_errors: Vec<ReplayError>,
41}
42
43pub(super) struct CassetteState {
44    runtime: Mutex<CassetteRuntime>,
45}
46
47impl CassetteState {
48    pub(super) fn new(cassette: HidCassette) -> Result<Arc<Self>, ReplayError> {
49        cassette.validate()?;
50        let mut queues: HashMap<_, VecDeque<_>> = HashMap::new();
51        let exchanges = cassette
52            .exchanges
53            .into_iter()
54            .enumerate()
55            .map(|(index, exchange)| {
56                let key = RequestKey::from_exchange(exchange.request_match, &exchange.request);
57                queues.entry(key.clone()).or_default().push_back(index);
58                PendingExchange {
59                    key,
60                    response: exchange.response,
61                    required: exchange.required,
62                    consumed: false,
63                }
64            })
65            .collect();
66        Ok(Arc::new(Self {
67            runtime: Mutex::new(CassetteRuntime {
68                exchanges,
69                queues,
70                unmatched: Vec::new(),
71                receiver_slots: ReceiverSlots::default(),
72                slot_state_errors: Vec::new(),
73            }),
74        }))
75    }
76
77    pub(super) fn declare_slots(&self, slots: Vec<ReceiverSlot>) -> Result<(), ReplayError> {
78        self.runtime
79            .lock()
80            .unwrap_or_else(PoisonError::into_inner)
81            .receiver_slots
82            .declare(slots)
83    }
84
85    pub(super) fn set_slot(&self, slot: u8, state: ReceiverSlotState) -> Result<(), ReplayError> {
86        self.runtime
87            .lock()
88            .unwrap_or_else(PoisonError::into_inner)
89            .receiver_slots
90            .set(slot, state)
91    }
92
93    pub(super) fn slot(&self, slot: u8) -> Result<ReceiverSlotState, ReplayError> {
94        self.runtime
95            .lock()
96            .unwrap_or_else(PoisonError::into_inner)
97            .receiver_slots
98            .get(slot)
99    }
100
101    pub(super) fn validate_report(&self, report: &[u8]) -> Result<(), ReplayError> {
102        let mut runtime = self.runtime.lock().unwrap_or_else(PoisonError::into_inner);
103        runtime.validate_slot_state(None, report)
104    }
105
106    fn respond(&self, actual: &[u8]) -> Result<Option<Vec<u8>>, ReplayError> {
107        let normalized = RequestMatch::Hidpp20.request_key(actual);
108        let exact = RequestKey::Exact(actual.to_vec());
109        let hidpp20 = RequestKey::Hidpp20(normalized.clone());
110        let mut runtime = self.runtime.lock().unwrap_or_else(PoisonError::into_inner);
111        let matched = runtime
112            .queues
113            .get(&exact)
114            .and_then(VecDeque::front)
115            .map(|&index| (index, RequestMatch::Exact))
116            .or_else(|| {
117                runtime
118                    .queues
119                    .get(&hidpp20)
120                    .and_then(VecDeque::front)
121                    .map(|&index| (index, RequestMatch::Hidpp20))
122            });
123        let Some((index, request_match)) = matched else {
124            let mismatch = ReplayMismatch {
125                actual: format_hex(actual),
126                normalized: format_hex(&normalized),
127            };
128            runtime.unmatched.push(mismatch.clone());
129            return Err(ReplayError::UnmatchedRequest {
130                actual: mismatch.actual,
131                normalized: mismatch.normalized,
132            });
133        };
134        let exchange = &runtime.exchanges[index];
135        let mut response = exchange.response.clone();
136        if request_match == RequestMatch::Hidpp20
137            && let Some(response) = response.as_mut()
138        {
139            rebind_software_id(response, actual[3] & 0x0f);
140        }
141        if let Some(response) = &response {
142            runtime.validate_slot_state(Some(actual), response)?;
143        }
144        // Validation and consumption share the slot-state lock. A rejected
145        // response remains at the head of its queue for a later valid state.
146        runtime
147            .queues
148            .get_mut(&RequestKey::from_exchange(request_match, actual))
149            .and_then(VecDeque::pop_front);
150        runtime.exchanges[index].consumed = true;
151        Ok(response)
152    }
153
154    pub(super) fn completion(&self, written_reports: Vec<Vec<u8>>) -> ReplayCompletion {
155        let runtime = self.runtime.lock().unwrap_or_else(PoisonError::into_inner);
156        let unconsumed_required = runtime
157            .exchanges
158            .iter()
159            .filter(|exchange| exchange.required && !exchange.consumed)
160            .map(|exchange| exchange.key.description())
161            .collect();
162        let consumed_optional = runtime
163            .exchanges
164            .iter()
165            .filter(|exchange| !exchange.required && exchange.consumed)
166            .count();
167        let unused_optional = runtime
168            .exchanges
169            .iter()
170            .filter(|exchange| !exchange.required && !exchange.consumed)
171            .count();
172        ReplayCompletion {
173            written_reports,
174            unmatched_requests: runtime.unmatched.clone(),
175            slot_state_errors: runtime.slot_state_errors.clone(),
176            unconsumed_required,
177            consumed_optional,
178            unused_optional,
179            channel_open_count: 0,
180        }
181    }
182}
183
184impl CassetteRuntime {
185    fn validate_slot_state(
186        &mut self,
187        request: Option<&[u8]>,
188        report: &[u8],
189    ) -> Result<(), ReplayError> {
190        self.receiver_slots
191            .validate(request, report)
192            .inspect_err(|error| {
193                self.slot_state_errors.push(error.clone());
194            })
195    }
196}
197
198fn rebind_software_id(response: &mut [u8], software_id: u8) {
199    if response[2] == 0xff {
200        response[4] = response[4] & 0xf0 | software_id;
201    } else {
202        response[3] = response[3] & 0xf0 | software_id;
203    }
204}
205
206/// One outgoing report that did not match a pending cassette exchange.
207#[derive(Clone, Debug, PartialEq, Eq)]
208pub struct ReplayMismatch {
209    /// Exact outgoing bytes as lowercase hex.
210    pub actual: String,
211    /// HID++ 2.0 candidate key with only the software-ID nibble cleared.
212    pub normalized: String,
213}
214
215/// Current cassette consumption and write-capture state.
216#[derive(Clone, Debug, PartialEq, Eq)]
217pub struct ReplayCompletion {
218    /// Every report written through the channel, in observed order.
219    pub written_reports: Vec<Vec<u8>>,
220    /// Outgoing reports that matched no pending exchange.
221    pub unmatched_requests: Vec<ReplayMismatch>,
222    /// Rejected evidence contradicting explicit receiver-slot state.
223    ///
224    /// These diagnostics remain even after recovery consumes the queued exchange.
225    pub slot_state_errors: Vec<ReplayError>,
226    /// Required normalized request keys that remain unused.
227    pub unconsumed_required: Vec<String>,
228    /// Number of optional exchanges that were consumed.
229    pub consumed_optional: usize,
230    /// Number of optional exchanges that remain unused.
231    pub unused_optional: usize,
232    /// Number of times a replay backend opened this logical channel.
233    ///
234    /// Standalone [`ReplayRawHidChannel`] handles report zero because no
235    /// backend open occurred.
236    pub channel_open_count: usize,
237}
238
239impl ReplayCompletion {
240    /// Whether no replay contract failed and every required exchange was consumed.
241    #[must_use]
242    pub fn is_complete(&self) -> bool {
243        self.unmatched_requests.is_empty()
244            && self.slot_state_errors.is_empty()
245            && self.unconsumed_required.is_empty()
246    }
247
248    /// Fail with the remaining required request keys when replay is incomplete.
249    pub fn require_complete(&self) -> Result<(), ReplayError> {
250        if let Some(error) = self.slot_state_errors.first() {
251            return Err(error.clone());
252        }
253        if let Some(mismatch) = self.unmatched_requests.first() {
254            return Err(ReplayError::UnmatchedRequest {
255                actual: mismatch.actual.clone(),
256                normalized: mismatch.normalized.clone(),
257            });
258        }
259        if self.unconsumed_required.is_empty() {
260            Ok(())
261        } else {
262            Err(ReplayError::UnconsumedExchanges {
263                requests: self.unconsumed_required.clone(),
264            })
265        }
266    }
267}
268
269/// Inspection and connection control for a [`ReplayRawHidChannel`].
270#[derive(Clone)]
271pub struct ReplayChannelHandle {
272    cassette: Option<Arc<CassetteState>>,
273    written: Arc<Mutex<Vec<Vec<u8>>>>,
274    connected: Arc<AtomicBool>,
275}
276
277impl ReplayChannelHandle {
278    pub(super) fn from_parts(
279        cassette: Arc<CassetteState>,
280        written: Arc<Mutex<Vec<Vec<u8>>>>,
281        connected: Arc<AtomicBool>,
282    ) -> Self {
283        Self {
284            cassette: Some(cassette),
285            written,
286            connected,
287        }
288    }
289
290    /// Every report written through this logical channel.
291    #[must_use]
292    pub fn written_reports(&self) -> Vec<Vec<u8>> {
293        self.written
294            .lock()
295            .unwrap_or_else(PoisonError::into_inner)
296            .clone()
297    }
298
299    /// Current strict cassette consumption and diagnostics.
300    #[must_use]
301    pub fn completion(&self) -> ReplayCompletion {
302        self.cassette.as_ref().map_or_else(
303            || ReplayCompletion {
304                written_reports: self.written_reports(),
305                unmatched_requests: Vec::new(),
306                slot_state_errors: Vec::new(),
307                unconsumed_required: Vec::new(),
308                consumed_optional: 0,
309                unused_optional: 0,
310                channel_open_count: 0,
311            },
312            |cassette| cassette.completion(self.written_reports()),
313        )
314    }
315
316    /// Fail unless every required exchange was consumed without a mismatch.
317    pub fn require_complete(&self) -> Result<(), ReplayError> {
318        self.completion().require_complete()
319    }
320
321    /// Change whether this channel lifetime accepts further writes.
322    pub fn set_connection(&self, connection: super::ChannelConnection) {
323        self.connected.store(
324            connection == super::ChannelConnection::Connected,
325            Ordering::SeqCst,
326        );
327    }
328}
329
330/// A strict cassette-backed implementation of the production raw HID contract.
331pub struct ReplayRawHidChannel {
332    vendor_id: u16,
333    product_id: u16,
334    report_support: ReportSupport,
335    incoming_tx: mpsc::UnboundedSender<Vec<u8>>,
336    incoming_rx: tokio::sync::Mutex<mpsc::UnboundedReceiver<Vec<u8>>>,
337    written: Arc<Mutex<Vec<Vec<u8>>>>,
338    cassette: Option<Arc<CassetteState>>,
339    response_gates: Arc<ResponseGates>,
340    responder: Option<Responder>,
341    connected: Arc<AtomicBool>,
342    fails: Option<fn(&[u8]) -> bool>,
343}
344
345impl ReplayRawHidChannel {
346    /// Build one connected raw channel over `cassette`.
347    pub fn new(
348        cassette: HidCassette,
349        vendor_id: u16,
350        product_id: u16,
351    ) -> Result<(Self, ReplayChannelHandle), ReplayError> {
352        let report_support = cassette.report_support;
353        let cassette = CassetteState::new(cassette)?;
354        let written = Arc::new(Mutex::new(Vec::new()));
355        let connected = Arc::new(AtomicBool::new(true));
356        let response_gates = Arc::new(ResponseGates::default());
357        let channel = Self::from_parts(
358            vendor_id,
359            product_id,
360            report_support,
361            Arc::clone(&cassette),
362            Arc::clone(&written),
363            Arc::clone(&connected),
364            response_gates,
365        );
366        let handle = ReplayChannelHandle::from_parts(cassette, written, connected);
367        Ok((channel, handle))
368    }
369
370    pub(super) fn from_parts(
371        vendor_id: u16,
372        product_id: u16,
373        report_support: ReportSupport,
374        cassette: Arc<CassetteState>,
375        written: Arc<Mutex<Vec<Vec<u8>>>>,
376        connected: Arc<AtomicBool>,
377        response_gates: Arc<ResponseGates>,
378    ) -> Self {
379        let (incoming_tx, incoming_rx) = mpsc::unbounded_channel();
380        Self {
381            vendor_id,
382            product_id,
383            report_support,
384            incoming_tx,
385            incoming_rx: tokio::sync::Mutex::new(incoming_rx),
386            written,
387            cassette: Some(cassette),
388            response_gates,
389            responder: None,
390            connected,
391            fails: None,
392        }
393    }
394
395    pub(super) fn incoming_sender(&self) -> mpsc::UnboundedSender<Vec<u8>> {
396        self.incoming_tx.clone()
397    }
398
399    #[cfg(test)]
400    pub(crate) fn with_responder(
401        responder: fn(&[u8]) -> Option<Vec<u8>>,
402    ) -> (Self, ReplayChannelHandle) {
403        Self::build_scripted(responder, None)
404    }
405
406    #[cfg(test)]
407    pub(crate) fn with_dynamic_responder(
408        responder: impl Fn(&[u8]) -> Option<Vec<u8>> + Send + Sync + 'static,
409    ) -> (Self, ReplayChannelHandle) {
410        Self::build_scripted(responder, None)
411    }
412
413    #[cfg(test)]
414    pub(crate) fn with_failing_writes(
415        responder: fn(&[u8]) -> Option<Vec<u8>>,
416        fails: fn(&[u8]) -> bool,
417    ) -> (Self, ReplayChannelHandle) {
418        Self::build_scripted(responder, Some(fails))
419    }
420
421    /// Present a scripted receiver's product id to HID++ receiver detection.
422    #[cfg(test)]
423    pub(crate) fn presenting_as(mut self, product_id: u16) -> Self {
424        self.product_id = product_id;
425        self
426    }
427
428    #[cfg(test)]
429    fn build_scripted(
430        responder: impl Fn(&[u8]) -> Option<Vec<u8>> + Send + Sync + 'static,
431        fails: Option<fn(&[u8]) -> bool>,
432    ) -> (Self, ReplayChannelHandle) {
433        let (incoming_tx, incoming_rx) = mpsc::unbounded_channel();
434        let written = Arc::new(Mutex::new(Vec::new()));
435        let connected = Arc::new(AtomicBool::new(true));
436        (
437            Self {
438                vendor_id: 0x046d,
439                product_id: 0xb35b,
440                report_support: ReportSupport::ShortAndLong,
441                incoming_tx,
442                incoming_rx: tokio::sync::Mutex::new(incoming_rx),
443                written: Arc::clone(&written),
444                cassette: None,
445                response_gates: Arc::new(ResponseGates::default()),
446                responder: Some(Arc::new(responder)),
447                connected: Arc::clone(&connected),
448                fails,
449            },
450            ReplayChannelHandle {
451                cassette: None,
452                written,
453                connected,
454            },
455        )
456    }
457}
458
459#[hidpp::async_trait]
460impl RawHidChannel for ReplayRawHidChannel {
461    fn vendor_id(&self) -> u16 {
462        self.vendor_id
463    }
464
465    fn product_id(&self) -> u16 {
466        self.product_id
467    }
468
469    async fn write_report(&self, src: &[u8]) -> Result<usize, Box<dyn Error + Send + Sync>> {
470        self.written
471            .lock()
472            .unwrap_or_else(PoisonError::into_inner)
473            .push(src.to_vec());
474        if !self.connected.load(Ordering::SeqCst) {
475            return Err(disconnected_error());
476        }
477        if self.fails.is_some_and(|fails| fails(src)) {
478            return Err(disconnected_error());
479        }
480        let response = if let Some(cassette) = &self.cassette {
481            cassette.respond(src)?
482        } else {
483            self.responder.as_ref().and_then(|responder| responder(src))
484        };
485        if let Some(gate) = self.response_gates.take(src) {
486            gate.request_written(self.incoming_tx.clone(), response);
487            return Ok(src.len());
488        }
489        if let Some(response) = response {
490            self.incoming_tx
491                .send(response)
492                .map_err(|_| disconnected_error())?;
493        }
494        Ok(src.len())
495    }
496
497    async fn read_report(&self, buf: &mut [u8]) -> Result<usize, Box<dyn Error + Send + Sync>> {
498        let Some(report) = self.incoming_rx.lock().await.recv().await else {
499            return Err(disconnected_error());
500        };
501        let len = report.len().min(buf.len());
502        buf[..len].copy_from_slice(&report[..len]);
503        Ok(len)
504    }
505
506    fn is_connected(&self) -> bool {
507        self.connected.load(Ordering::SeqCst)
508    }
509
510    fn supports_short_long_hidpp(&self) -> Option<(bool, bool)> {
511        Some((
512            self.report_support.supports_short_reports(),
513            self.report_support.supports_long_reports(),
514        ))
515    }
516
517    async fn get_report_descriptor(
518        &self,
519        _buf: &mut [u8],
520    ) -> Result<usize, Box<dyn Error + Send + Sync>> {
521        unreachable!("replay channel declares HID++ report support")
522    }
523}
524
525fn disconnected_error() -> Box<dyn Error + Send + Sync> {
526    Box::new(io::Error::new(
527        io::ErrorKind::BrokenPipe,
528        "replay HID channel is disconnected",
529    ))
530}
531
532/// A raw output-report sink that records every successful write.
533pub struct ReplayRawWriter {
534    written: Arc<Mutex<Vec<Vec<u8>>>>,
535    connected: Arc<AtomicBool>,
536}
537
538impl ReplayRawWriter {
539    /// Build one connected recording writer and its inspection handle.
540    #[must_use]
541    pub fn new() -> (Self, ReplayRawWriterHandle) {
542        let written = Arc::new(Mutex::new(Vec::new()));
543        let connected = Arc::new(AtomicBool::new(true));
544        (
545            Self {
546                written: Arc::clone(&written),
547                connected: Arc::clone(&connected),
548            },
549            ReplayRawWriterHandle { written, connected },
550        )
551    }
552
553    pub(super) fn from_parts(
554        written: Arc<Mutex<Vec<Vec<u8>>>>,
555        connected: Arc<AtomicBool>,
556    ) -> Self {
557        Self { written, connected }
558    }
559}
560
561#[hidpp::async_trait]
562impl RawWriter for ReplayRawWriter {
563    async fn write_output_report(&mut self, report: &[u8]) -> Result<(), BackendError> {
564        if !self.connected.load(Ordering::SeqCst) {
565            return Err(BackendError::Disconnected);
566        }
567        self.written
568            .lock()
569            .unwrap_or_else(PoisonError::into_inner)
570            .push(report.to_vec());
571        Ok(())
572    }
573}
574
575/// Inspection and connection control for a [`ReplayRawWriter`].
576#[derive(Clone)]
577pub struct ReplayRawWriterHandle {
578    written: Arc<Mutex<Vec<Vec<u8>>>>,
579    connected: Arc<AtomicBool>,
580}
581
582impl ReplayRawWriterHandle {
583    pub(super) fn from_parts(
584        written: Arc<Mutex<Vec<Vec<u8>>>>,
585        connected: Arc<AtomicBool>,
586    ) -> Self {
587        Self { written, connected }
588    }
589
590    /// Every output report accepted by this writer.
591    #[must_use]
592    pub fn written_reports(&self) -> Vec<Vec<u8>> {
593        self.written
594            .lock()
595            .unwrap_or_else(PoisonError::into_inner)
596            .clone()
597    }
598
599    /// Change whether the writer accepts further reports.
600    pub fn set_connection(&self, connection: super::ChannelConnection) {
601        self.connected.store(
602            connection == super::ChannelConnection::Connected,
603            Ordering::SeqCst,
604        );
605    }
606}
openlogi_device/replay/channel.rs

openlogi_device/replay/
channel.rs
