Skip to main content

spvirit_server/
handler.rs

1//! PVA protocol handler — the core TCP connection processor.
2//!
3//! [`handle_connection`] uses a [`SourceRegistry`] to resolve PV names across
4//! multiple registered sources, serving PVs over the EPICS PVAccess protocol.
5
6use std::net::{IpAddr, Ipv4Addr, SocketAddr};
7use std::sync::Arc;
8use std::sync::atomic::{AtomicU16, AtomicU32, Ordering};
9use std::time::{Duration, Instant, SystemTime};
10
11use regex::Regex;
12use tokio::io::{AsyncReadExt, AsyncWriteExt};
13use tokio::net::{TcpListener, TcpStream, UdpSocket};
14use tokio::sync::mpsc;
15use tracing::{debug, error, info};
16
17use spvirit_codec::epics_decode::{PvaHeader, PvaPacket, PvaPacketCommand};
18use spvirit_codec::spvd_decode::{PvdDecoder, StructureDesc, extract_subfield_desc};
19use spvirit_codec::spvd_encode::{
20    decode_pv_request_fields, filter_structure_desc, nt_payload_desc,
21};
22use spvirit_codec::spvirit_encode::{
23    encode_connection_validation, encode_control_message, encode_create_channel_error,
24    encode_create_channel_response, encode_get_field_error, encode_get_field_response,
25    encode_header, encode_message_error, encode_monitor_data_response_payload,
26    encode_op_data_response_filtered, encode_op_error, encode_op_get_data_response_payload,
27    encode_op_init_response_desc, encode_op_put_get_data_error_response,
28    encode_op_put_get_data_response_payload, encode_op_put_get_init_error_response,
29    encode_op_put_get_init_response, encode_op_put_getput_response_payload, encode_op_put_response,
30    encode_op_put_status_response, encode_op_rpc_data_response_payload,
31    encode_op_status_error_response, encode_op_status_response, encode_search_response,
32    ip_from_bytes, ip_to_bytes,
33};
34
35use spvirit_types::{NtPayload, NtScalar, NtScalarArray, ScalarArrayValue, ScalarValue};
36
37use crate::decode::{assemble_segmented_message, decode_put_body};
38use crate::monitor::MonitorRegistry;
39use crate::pvstore::SourceRegistry;
40use crate::state::{ConnState, MonitorState, MonitorSub};
41
42// ---------------------------------------------------------------------------
43// PvListMode — controls virtual PV listing behaviour
44// ---------------------------------------------------------------------------
45
46/// Controls how the server exposes its PV directory.
47#[derive(Debug, Clone, Copy, PartialEq, Eq)]
48pub enum PvListMode {
49    /// No PV listing at all.
50    Off,
51    /// Respond to UDP search for known PVs only; no GET_FIELD listing.
52    Discover,
53    /// Full pvlist & server-RPC listing support.
54    List,
55}
56
57impl PvListMode {
58    pub fn parse(raw: &str) -> Result<Self, String> {
59        match raw.trim().to_ascii_lowercase().as_str() {
60            "off" => Ok(Self::Off),
61            "discover" => Ok(Self::Discover),
62            "list" => Ok(Self::List),
63            other => Err(format!(
64                "Invalid pvlist-mode '{}'; expected off|discover|list",
65                other
66            )),
67        }
68    }
69}
70
71// ---------------------------------------------------------------------------
72// Server shared state
73// ---------------------------------------------------------------------------
74
75/// Shared server state that is passed to every connection handler.
76pub struct ServerState {
77    pub sources: Arc<SourceRegistry>,
78    pub registry: Arc<MonitorRegistry>,
79    pub sid_counter: AtomicU32,
80    pub beacon_change: Arc<AtomicU16>,
81    pub compute_alarms: bool,
82    pub pvlist_mode: PvListMode,
83    pub pvlist_max: usize,
84    pub pvlist_allow_pattern: Option<Regex>,
85    pub guid: [u8; 12],
86    pub tcp_port: u16,
87    pub advertise_ip: Option<IpAddr>,
88    pub listen_ip: IpAddr,
89}
90
91impl ServerState {
92    pub fn new(
93        sources: Arc<SourceRegistry>,
94        registry: Arc<MonitorRegistry>,
95        compute_alarms: bool,
96        pvlist_mode: PvListMode,
97        pvlist_max: usize,
98        pvlist_allow_pattern: Option<Regex>,
99        guid: [u8; 12],
100        tcp_port: u16,
101        advertise_ip: Option<IpAddr>,
102        listen_ip: IpAddr,
103    ) -> Self {
104        Self {
105            sources,
106            registry,
107            sid_counter: AtomicU32::new(1),
108            beacon_change: Arc::new(AtomicU16::new(0)),
109            compute_alarms,
110            pvlist_mode,
111            pvlist_max,
112            pvlist_allow_pattern,
113            guid,
114            tcp_port,
115            advertise_ip,
116            listen_ip,
117        }
118    }
119}
120
121// ---------------------------------------------------------------------------
122// Virtual PV helpers
123// ---------------------------------------------------------------------------
124
125pub fn is_pvlist_virtual_pv(pv_name: &str) -> bool {
126    pv_name == "__pvlist"
127}
128
129pub fn is_server_rpc_pv(pv_name: &str) -> bool {
130    pv_name == "server"
131}
132
133pub fn is_virtual_event_pv(pv_name: &str) -> bool {
134    pv_name.starts_with("__event:")
135}
136
137pub fn virtual_event_nt(pv_name: &str) -> NtPayload {
138    NtPayload::Scalar(
139        NtScalar::from_value(ScalarValue::Bool(false))
140            .with_description(format!("Virtual event trigger for {}", pv_name)),
141    )
142}
143
144pub fn virtual_pvlist_nt(entries: Vec<String>) -> NtPayload {
145    NtPayload::ScalarArray(NtScalarArray::from_value(ScalarArrayValue::Str(entries)))
146}
147
148// ---------------------------------------------------------------------------
149// Pattern / wildcard utilities
150// ---------------------------------------------------------------------------
151
152pub fn is_pattern_query(raw: &str) -> bool {
153    raw.contains('*') || raw.contains('?')
154}
155
156pub fn wildcard_match(pattern: &str, text: &str) -> bool {
157    let p = pattern.as_bytes();
158    let t = text.as_bytes();
159    let mut i = 0usize;
160    let mut j = 0usize;
161    let mut star: Option<usize> = None;
162    let mut match_j = 0usize;
163
164    while j < t.len() {
165        if i < p.len() && (p[i] == b'?' || p[i] == t[j]) {
166            i += 1;
167            j += 1;
168        } else if i < p.len() && p[i] == b'*' {
169            star = Some(i);
170            i += 1;
171            match_j = j;
172        } else if let Some(star_idx) = star {
173            i = star_idx + 1;
174            match_j += 1;
175            j = match_j;
176        } else {
177            return false;
178        }
179    }
180
181    while i < p.len() && p[i] == b'*' {
182        i += 1;
183    }
184    i == p.len()
185}
186
187pub fn collect_visible_pv_names(
188    all_names: &[String],
189    mode: PvListMode,
190    allow_pattern: Option<&Regex>,
191    max_items: usize,
192) -> Vec<String> {
193    let mut names: Vec<String> = all_names
194        .iter()
195        .filter(|name| {
196            allow_pattern
197                .as_ref()
198                .map(|re| re.is_match(name))
199                .unwrap_or(true)
200        })
201        .cloned()
202        .collect();
203    names.sort();
204    if names.len() > max_items {
205        names.truncate(max_items);
206    }
207    if mode == PvListMode::List && names.len() < max_items {
208        names.push("__pvlist".to_string());
209    }
210    names
211}
212
213fn build_pvlist_structure(names: &[String]) -> StructureDesc {
214    use spvirit_codec::spvd_decode::{FieldDesc, FieldType, TypeCode};
215    StructureDesc {
216        struct_id: Some("epics:pva/pvlist:1.0".to_string()),
217        fields: names
218            .iter()
219            .map(|name| FieldDesc {
220                name: name.clone(),
221                field_type: FieldType::Scalar(TypeCode::Boolean),
222            })
223            .collect(),
224    }
225}
226
227fn requested_pvlist_pattern(field_name: Option<&str>) -> Option<&str> {
228    let raw = field_name.map(str::trim).unwrap_or("");
229    if raw.is_empty() || raw == "*" || raw == "__pvlist" || raw.eq_ignore_ascii_case("pvlist") {
230        return Some("*");
231    }
232    if is_pattern_query(raw) {
233        return Some(raw);
234    }
235    None
236}
237
238// ---------------------------------------------------------------------------
239// Network helpers
240// ---------------------------------------------------------------------------
241
242pub fn search_reply_target(addr: &[u8; 16], port: u16, peer: SocketAddr) -> SocketAddr {
243    let target_port = if port != 0 { port } else { peer.port() };
244    let target_ip = ip_from_bytes(addr)
245        .filter(|ip| !ip.is_unspecified())
246        .unwrap_or_else(|| peer.ip());
247    SocketAddr::new(target_ip, target_port)
248}
249
250/// Bind the fixed UDP search port with `SO_REUSEADDR` (and `SO_REUSEPORT`
251/// on Unix) so other local PVA consumers such as `p4p` can also listen on
252/// the same well-known port. On macOS in particular, a plain
253/// `UdpSocket::bind(5076)` prevents any subsequent binder from joining the
254/// port, which broke co-located clients.
255pub fn bind_udp_search_socket(addr: SocketAddr) -> std::io::Result<UdpSocket> {
256    use socket2::{Domain, Protocol, Socket, Type};
257
258    let domain = if addr.is_ipv4() {
259        Domain::IPV4
260    } else {
261        Domain::IPV6
262    };
263    let socket = Socket::new(domain, Type::DGRAM, Some(Protocol::UDP))?;
264    socket.set_reuse_address(true)?;
265    #[cfg(unix)]
266    socket.set_reuse_port(true)?;
267    socket.set_nonblocking(true)?;
268    socket.bind(&addr.into())?;
269    UdpSocket::from_std(socket.into())
270}
271
272pub fn infer_udp_response_ip(peer: SocketAddr) -> Option<IpAddr> {
273    let bind_addr = if peer.is_ipv4() {
274        "0.0.0.0:0"
275    } else {
276        "[::]:0"
277    };
278    let sock = std::net::UdpSocket::bind(bind_addr).ok()?;
279    sock.connect(peer).ok()?;
280    let local = sock.local_addr().ok()?;
281    if local.ip().is_unspecified() {
282        None
283    } else {
284        Some(local.ip())
285    }
286}
287
288pub fn rand_guid() -> [u8; 12] {
289    let now = SystemTime::now()
290        .duration_since(SystemTime::UNIX_EPOCH)
291        .unwrap_or_default();
292    let mut guid = [0u8; 12];
293    let bytes = now.as_nanos().to_le_bytes();
294    guid.copy_from_slice(&bytes[0..12]);
295    guid
296}
297
298// ---------------------------------------------------------------------------
299// Debug utilities
300// ---------------------------------------------------------------------------
301
302pub fn validate_encoded_packet(conn_id: u64, label: &str, bytes: &[u8]) {
303    let mut pkt = PvaPacket::new(bytes);
304    let decoded = pkt.decode_payload();
305    match decoded {
306        Some(PvaPacketCommand::ConnectionValidation(payload)) => {
307            debug!(
308                "Conn {}: {} decoded as cmd=1 buffer_size={} qos={} authz={:?}",
309                conn_id, label, payload.buffer_size, payload.qos, payload.authz
310            );
311        }
312        Some(PvaPacketCommand::ConnectionValidated(_)) => {
313            debug!("Conn {}: {} decoded as cmd=9", conn_id, label);
314        }
315        Some(other) => {
316            debug!("Conn {}: {} decoded as {:?}", conn_id, label, other);
317        }
318        None => {
319            debug!("Conn {}: {} failed to decode", conn_id, label);
320        }
321    }
322}
323
324pub fn dump_hex_packet(
325    conn_id: u64,
326    dir: &str,
327    label: &str,
328    version: u8,
329    is_be: bool,
330    bytes: &[u8],
331) {
332    debug!(
333        "Conn {}: {} {} ver={} be={} len={}",
334        conn_id,
335        dir,
336        label,
337        version,
338        is_be,
339        bytes.len()
340    );
341    let mut offset = 0usize;
342    while offset < bytes.len() {
343        let end = usize::min(offset + 16, bytes.len());
344        let chunk = &bytes[offset..end];
345        let mut line = String::new();
346        for (i, b) in chunk.iter().enumerate() {
347            if i > 0 {
348                line.push(' ');
349            }
350            line.push_str(&format!("{:02x}", b));
351        }
352        debug!("Conn {}: {:04x} {}", conn_id, offset, line);
353        offset += 16;
354    }
355}
356
357// ---------------------------------------------------------------------------
358// Store-based snapshot/writable helpers (delegate to SourceRegistry + virtual PVs)
359// ---------------------------------------------------------------------------
360
361async fn get_nt_snapshot(state: &ServerState, pv_name: &str) -> Option<NtPayload> {
362    if is_pvlist_virtual_pv(pv_name) {
363        if state.pvlist_mode != PvListMode::List {
364            return None;
365        }
366        let all_names = state.sources.names().await;
367        let names = collect_visible_pv_names(
368            &all_names,
369            state.pvlist_mode,
370            state.pvlist_allow_pattern.as_ref(),
371            state.pvlist_max,
372        );
373        return Some(virtual_pvlist_nt(names));
374    }
375    if is_virtual_event_pv(pv_name) {
376        return Some(virtual_event_nt(pv_name));
377    }
378    state.sources.get(pv_name).await
379}
380
381async fn is_writable_pv(state: &ServerState, pv_name: &str) -> bool {
382    if is_virtual_event_pv(pv_name) {
383        return true;
384    }
385    state.sources.is_writable(pv_name).await
386}
387
388async fn has_pv(state: &ServerState, pv_name: &str) -> bool {
389    state.sources.has_pv(pv_name).await
390        || is_virtual_event_pv(pv_name)
391        || (is_pvlist_virtual_pv(pv_name) && state.pvlist_mode == PvListMode::List)
392        || (is_server_rpc_pv(pv_name) && state.pvlist_mode != PvListMode::Off)
393}
394
395// ---------------------------------------------------------------------------
396// Notify helpers
397// ---------------------------------------------------------------------------
398
399async fn notify_changed_records(
400    state: &ServerState,
401    changed: Vec<(String, NtPayload)>,
402) {
403    for (name, payload) in changed {
404        state.beacon_change.fetch_add(1, Ordering::SeqCst);
405        state.registry.notify_monitors(&name, &payload).await;
406    }
407}
408
409// ---------------------------------------------------------------------------
410// GET_FIELD handler
411// ---------------------------------------------------------------------------
412
413async fn handle_get_field_request(
414    state: &ServerState,
415    conn_state: &ConnState,
416    conn_id: u64,
417    payload: spvirit_codec::epics_decode::PvaGetFieldPayload,
418    version: u8,
419    is_be: bool,
420) {
421    if payload.is_server {
422        let resp = encode_get_field_error(
423            payload.cid,
424            "Unexpected server GET_FIELD payload",
425            version,
426            is_be,
427        );
428        state.registry.send_msg(conn_id, resp).await;
429        return;
430    }
431
432    let request_id = payload.ioid.unwrap_or(payload.cid);
433
434    let sid = payload
435        .sid
436        .or_else(|| conn_state.cid_to_sid.get(&payload.cid).copied())
437        .or_else(|| {
438            conn_state
439                .sid_to_pv
440                .contains_key(&payload.cid)
441                .then_some(payload.cid)
442        })
443        .or_else(|| {
444            (payload.cid == 0 && conn_state.sid_to_pv.len() == 1)
445                .then(|| conn_state.sid_to_pv.keys().copied().next())
446                .flatten()
447        });
448
449    if let Some(sid) = sid {
450        if let Some(pv_name) = conn_state.sid_to_pv.get(&sid) {
451            if let Some(nt) = get_nt_snapshot(state, pv_name).await {
452                let full_desc = nt_payload_desc(&nt);
453                let sub = payload.field_name.as_deref().filter(|s| !s.is_empty());
454                let desc = if let Some(field_path) = sub {
455                    match extract_subfield_desc(&full_desc, field_path) {
456                        Some(sub_desc) => sub_desc,
457                        None => {
458                            let resp = encode_get_field_error(
459                                request_id,
460                                &format!("sub-field '{}' not found", field_path),
461                                version,
462                                is_be,
463                            );
464                            state.registry.send_msg(conn_id, resp).await;
465                            return;
466                        }
467                    }
468                } else {
469                    full_desc
470                };
471                let resp = encode_get_field_response(request_id, &desc, version, is_be);
472                dump_hex_packet(conn_id, "tx", "cmd=17 get_field", version, is_be, &resp);
473                state.registry.send_msg(conn_id, resp).await;
474                debug!(
475                    "Conn {}: get_field cid={} sid={:?} ioid={:?} resolved_sid={} pv='{}' field={:?}",
476                    conn_id,
477                    payload.cid,
478                    payload.sid,
479                    payload.ioid,
480                    sid,
481                    pv_name,
482                    payload.field_name
483                );
484                return;
485            }
486            let resp = encode_get_field_error(request_id, "PV not found", version, is_be);
487            state.registry.send_msg(conn_id, resp).await;
488            return;
489        }
490    }
491
492    if state.pvlist_mode != PvListMode::List {
493        let resp = encode_get_field_error(
494            request_id,
495            "GET_FIELD listing is disabled (set --pvlist-mode=list)",
496            version,
497            is_be,
498        );
499        state.registry.send_msg(conn_id, resp).await;
500        return;
501    }
502
503    let Some(pattern) = requested_pvlist_pattern(payload.field_name.as_deref()) else {
504        let resp = encode_get_field_error(
505            request_id,
506            "GET_FIELD requires a valid list pattern",
507            version,
508            is_be,
509        );
510        state.registry.send_msg(conn_id, resp).await;
511        return;
512    };
513
514    let all_names = state.sources.names().await;
515    let mut names = collect_visible_pv_names(
516        &all_names,
517        state.pvlist_mode,
518        state.pvlist_allow_pattern.as_ref(),
519        state.pvlist_max,
520    );
521    if pattern != "*" {
522        names.retain(|name| wildcard_match(pattern, name));
523    }
524    if names.is_empty() {
525        let resp =
526            encode_get_field_error(request_id, "No PVs matched list request", version, is_be);
527        state.registry.send_msg(conn_id, resp).await;
528        return;
529    }
530    let desc = build_pvlist_structure(&names);
531    let resp = encode_get_field_response(request_id, &desc, version, is_be);
532    dump_hex_packet(
533        conn_id,
534        "tx",
535        "cmd=17 get_field_list",
536        version,
537        is_be,
538        &resp,
539    );
540    state.registry.send_msg(conn_id, resp).await;
541    debug!(
542        "Conn {}: get_field list pattern='{}' returned {} entries",
543        conn_id,
544        pattern,
545        names.len()
546    );
547}
548
549// ---------------------------------------------------------------------------
550// Server RPC handler
551// ---------------------------------------------------------------------------
552
553async fn handle_server_rpc(
554    state: &ServerState,
555    conn_id: u64,
556    ioid: u32,
557    subcmd: u8,
558    version: u8,
559    is_be: bool,
560) {
561    if state.pvlist_mode != PvListMode::List {
562        let resp = encode_op_status_error_response(
563            20,
564            ioid,
565            subcmd,
566            "RPC list endpoint disabled (set --pvlist-mode=list)",
567            version,
568            is_be,
569        );
570        state.registry.send_msg(conn_id, resp).await;
571        return;
572    }
573
574    let all_names = state.sources.names().await;
575    let names = collect_visible_pv_names(
576        &all_names,
577        state.pvlist_mode,
578        state.pvlist_allow_pattern.as_ref(),
579        state.pvlist_max,
580    );
581    let payload = NtPayload::ScalarArray(NtScalarArray::from_value(ScalarArrayValue::Str(names)));
582
583    let is_init = (subcmd & 0x08) != 0;
584    if is_init {
585        let resp = encode_op_status_response(20, ioid, subcmd, version, is_be);
586        state.registry.send_msg(conn_id, resp).await;
587        return;
588    }
589
590    let resp = encode_op_rpc_data_response_payload(ioid, subcmd, &payload, version, is_be);
591    state.registry.send_msg(conn_id, resp).await;
592}
593
594// ---------------------------------------------------------------------------
595// Control message handler (inside segmented stream)
596// ---------------------------------------------------------------------------
597
598async fn handle_control_message(
599    state: &ServerState,
600    conn_id: u64,
601    header: &PvaHeader,
602) {
603    debug!(
604        "Conn {}: control (segmented) cmd={} data={}",
605        conn_id, header.command, header.payload_length
606    );
607    if header.command == 3 {
608        let resp = encode_control_message(
609            true,
610            header.flags.is_msb,
611            header.version,
612            4,
613            header.payload_length,
614        );
615        state.registry.send_msg(conn_id, resp).await;
616    }
617}
618
619// ---------------------------------------------------------------------------
620// UDP search handler
621// ---------------------------------------------------------------------------
622
623/// Run the UDP search responder.
624pub async fn run_udp_search(
625    state: Arc<ServerState>,
626    addr: SocketAddr,
627    tcp_port: u16,
628    guid: [u8; 12],
629    advertise_ip: Option<IpAddr>,
630) -> Result<(), Box<dyn std::error::Error>> {
631    let socket = bind_udp_search_socket(addr)?;
632    socket.set_broadcast(true)?;
633    let mut buf = vec![0u8; 4096];
634
635    loop {
636        let (len, peer) = socket.recv_from(&mut buf).await?;
637        let data = &buf[..len];
638        let header = PvaHeader::new(data);
639        if header.flags.is_control || header.command != 3 {
640            continue;
641        }
642        let mut pkt = PvaPacket::new(data);
643        let Some(cmd) = pkt.decode_payload() else {
644            continue;
645        };
646        let version = pkt.header.version;
647        let is_be = pkt.header.flags.is_msb;
648        match cmd {
649            PvaPacketCommand::Search(payload) => {
650                debug!(
651                    "UDP search from {}: pv_count={} mask=0x{:02x}",
652                    peer,
653                    payload.pv_requests.len(),
654                    payload.mask
655                );
656                let accepts_tcp = payload.protocols.is_empty()
657                    || payload
658                        .protocols
659                        .iter()
660                        .any(|p| p.eq_ignore_ascii_case("tcp"));
661                if !accepts_tcp {
662                    debug!("UDP search: no compatible protocol (tcp not accepted)");
663                    continue;
664                }
665                let all_names = state.sources.names().await;
666                let visible_names = collect_visible_pv_names(
667                    &all_names,
668                    state.pvlist_mode,
669                    state.pvlist_allow_pattern.as_ref(),
670                    state.pvlist_max,
671                );
672                let mut cids = Vec::new();
673                for (cid, name) in &payload.pv_requests {
674                    if state.sources.has_pv(name).await
675                        || is_virtual_event_pv(name)
676                        || (is_pvlist_virtual_pv(name) && state.pvlist_mode == PvListMode::List)
677                        || (is_server_rpc_pv(name) && state.pvlist_mode != PvListMode::Off)
678                    {
679                        cids.push(*cid);
680                        continue;
681                    }
682                    if state.pvlist_mode != PvListMode::Off
683                        && is_pattern_query(name)
684                        && visible_names.iter().any(|pv| wildcard_match(name, pv))
685                    {
686                        cids.push(*cid);
687                    }
688                }
689                let response_required = (payload.mask & 0x01) != 0;
690                let server_discovery_ping = payload.pv_requests.is_empty();
691                let found = server_discovery_ping || !cids.is_empty();
692                if !found && !response_required {
693                    debug!("UDP search: no matches and response not required");
694                    continue;
695                }
696                let resp_ip = if let Some(ip) = advertise_ip {
697                    ip
698                } else if !addr.ip().is_unspecified() {
699                    addr.ip()
700                } else if let Some(ip) = infer_udp_response_ip(peer) {
701                    debug!("UDP search: inferred response address {}", ip);
702                    ip
703                } else {
704                    IpAddr::V4(Ipv4Addr::UNSPECIFIED)
705                };
706                let addr_bytes = if resp_ip.is_unspecified() {
707                    debug!("UDP search: responding with zero address (unspecified listen)");
708                    [0u8; 16]
709                } else {
710                    ip_to_bytes(resp_ip)
711                };
712                let response = encode_search_response(
713                    guid,
714                    payload.seq,
715                    addr_bytes,
716                    tcp_port,
717                    "tcp",
718                    found,
719                    &cids,
720                    version,
721                    is_be,
722                );
723                let reply_target = search_reply_target(&payload.addr, payload.port, peer);
724                if let Err(e) = socket.send_to(&response, reply_target).await {
725                    debug!(
726                        "UDP search: failed sending {} matches to {}: {}",
727                        cids.len(),
728                        reply_target,
729                        e
730                    );
731                    continue;
732                }
733                debug!(
734                    "UDP search: responded found={} with {} matches to {}",
735                    found,
736                    cids.len(),
737                    reply_target
738                );
739            }
740            _ => {}
741        }
742    }
743}
744
745// ---------------------------------------------------------------------------
746// TCP server
747// ---------------------------------------------------------------------------
748
749/// Accept TCP connections and spawn a handler for each.
750pub async fn run_tcp_server(
751    state: Arc<ServerState>,
752    addr: SocketAddr,
753    conn_timeout: Duration,
754) -> Result<(), Box<dyn std::error::Error>> {
755    let listener = TcpListener::bind(addr).await?;
756    let conn_id = Arc::new(std::sync::atomic::AtomicU64::new(1));
757
758    loop {
759        let (stream, peer) = listener.accept().await?;
760        let id = conn_id.fetch_add(1, Ordering::SeqCst);
761        info!("TCP connection {} from {}", id, peer);
762        let state_clone = state.clone();
763        tokio::spawn(async move {
764            if let Err(e) = handle_connection(state_clone, stream, id, conn_timeout).await {
765                error!("Connection {} error: {}", id, e);
766            }
767        });
768    }
769}
770
771// ---------------------------------------------------------------------------
772// Core TCP connection handler
773// ---------------------------------------------------------------------------
774
775/// Handle a single PVA TCP connection.
776///
777/// This is the main protocol loop: handshake, then dispatch each command
778/// (CreateChannel, GET, PUT, PUT_GET, MONITOR, RPC, etc.) using the
779/// [`SourceRegistry`] abstraction.
780pub async fn handle_connection(
781    state: Arc<ServerState>,
782    stream: TcpStream,
783    conn_id: u64,
784    conn_timeout: Duration,
785) -> Result<(), Box<dyn std::error::Error>> {
786    let (mut reader, mut writer) = stream.into_split();
787    let (tx, mut rx) = mpsc::channel::<Vec<u8>>(128);
788
789    {
790        let mut conns = state.registry.conns.lock().await;
791        conns.insert(conn_id, tx);
792    }
793
794    let writer_task = tokio::spawn(async move {
795        while let Some(msg) = rx.recv().await {
796            if writer.write_all(&msg).await.is_err() {
797                break;
798            }
799        }
800    });
801
802    let mut conn_state = ConnState::default();
803
804    // Per EPICS PVA protocol: send SET_BYTE_ORDER control message before validation.
805    let set_byte_order = encode_control_message(true, false, 2, 2, 0);
806    validate_encoded_packet(conn_id, "set_byte_order", &set_byte_order);
807    dump_hex_packet(
808        conn_id,
809        "tx",
810        "ctrl=2 set_byte_order",
811        2,
812        false,
813        &set_byte_order,
814    );
815    state.registry.send_msg(conn_id, set_byte_order).await;
816
817    // Server sends Connection Validation (cmd=1) next.
818    let server_validation = encode_connection_validation(16_384, 512, &["anonymous", "ca"], 2, false);
819    validate_encoded_packet(conn_id, "server_validation_init", &server_validation);
820    dump_hex_packet(
821        conn_id,
822        "tx",
823        "cmd=1 server_validation_init",
824        2,
825        false,
826        &server_validation,
827    );
828    state.registry.send_msg(conn_id, server_validation).await;
829
830    let mut last_activity = Instant::now();
831
832    loop {
833        let mut header = [0u8; 8];
834        let elapsed = last_activity.elapsed();
835        if elapsed >= conn_timeout {
836            info!("Conn {} idle timeout", conn_id);
837            break;
838        }
839        let remaining = conn_timeout - elapsed;
840        let read_header = tokio::time::timeout(remaining, reader.read_exact(&mut header)).await;
841        match read_header {
842            Ok(Ok(_)) => {}
843            Ok(Err(_)) => break,
844            Err(_) => {
845                info!("Conn {} idle timeout", conn_id);
846                break;
847            }
848        }
849        let header_pkt = PvaPacket::new(&header);
850        let payload_len = if header_pkt.header.flags.is_control {
851            0usize
852        } else {
853            header_pkt.header.payload_length as usize
854        };
855        let mut payload = vec![0u8; payload_len];
856        if payload_len > 0 {
857            let elapsed = last_activity.elapsed();
858            if elapsed >= conn_timeout {
859                info!("Conn {} idle timeout", conn_id);
860                break;
861            }
862            let remaining = conn_timeout - elapsed;
863            let read_payload =
864                tokio::time::timeout(remaining, reader.read_exact(&mut payload)).await;
865            match read_payload {
866                Ok(Ok(_)) => {}
867                Ok(Err(_)) => break,
868                Err(_) => {
869                    info!("Conn {} idle timeout", conn_id);
870                    break;
871                }
872            }
873        }
874        last_activity = Instant::now();
875        let mut full = header.to_vec();
876        full.extend_from_slice(&payload);
877
878        // Segmented message reassembly
879        if header_pkt.header.flags.is_segmented != 0 && !header_pkt.header.flags.is_control {
880            debug!(
881                "Conn {}: segmented message cmd={} seg=0x{:02x}",
882                conn_id, header_pkt.header.command, header_pkt.header.flags.is_segmented
883            );
884            let mut payloads = vec![payload];
885            let mut seg_flags = header_pkt.header.flags;
886            while !seg_flags.is_last_segment {
887                let mut seg_header = [0u8; 8];
888                let elapsed = last_activity.elapsed();
889                if elapsed >= conn_timeout {
890                    info!("Conn {} idle timeout", conn_id);
891                    break;
892                }
893                let remaining = conn_timeout - elapsed;
894                let read_header =
895                    tokio::time::timeout(remaining, reader.read_exact(&mut seg_header)).await;
896                match read_header {
897                    Ok(Ok(_)) => {}
898                    Ok(Err(_)) => break,
899                    Err(_) => {
900                        info!("Conn {} idle timeout", conn_id);
901                        break;
902                    }
903                }
904
905                let seg_header_pkt = PvaPacket::new(&seg_header);
906                let seg_payload_len = if seg_header_pkt.header.flags.is_control {
907                    0usize
908                } else {
909                    seg_header_pkt.header.payload_length as usize
910                };
911                let mut seg_payload = vec![0u8; seg_payload_len];
912                if seg_payload_len > 0 {
913                    let elapsed = last_activity.elapsed();
914                    if elapsed >= conn_timeout {
915                        info!("Conn {} idle timeout", conn_id);
916                        break;
917                    }
918                    let remaining = conn_timeout - elapsed;
919                    let read_payload =
920                        tokio::time::timeout(remaining, reader.read_exact(&mut seg_payload)).await;
921                    match read_payload {
922                        Ok(Ok(_)) => {}
923                        Ok(Err(_)) => break,
924                        Err(_) => {
925                            info!("Conn {} idle timeout", conn_id);
926                            break;
927                        }
928                    }
929                }
930                last_activity = Instant::now();
931
932                if seg_header_pkt.header.flags.is_control {
933                    handle_control_message(&state, conn_id, &seg_header_pkt.header).await;
934                    continue;
935                }
936                if seg_header_pkt.header.flags.is_segmented == 0 {
937                    debug!(
938                        "Conn {}: segmented message interrupted by non-segmented cmd={}",
939                        conn_id, seg_header_pkt.header.command
940                    );
941                    break;
942                }
943                payloads.push(seg_payload);
944                seg_flags = seg_header_pkt.header.flags;
945            }
946            full = assemble_segmented_message(header, payloads);
947        }
948
949        let mut pkt = PvaPacket::new(&full);
950        let Some(cmd) = pkt.decode_payload() else {
951            continue;
952        };
953        let version = pkt.header.version;
954        let is_be = pkt.header.flags.is_msb;
955        let cmd_code = pkt.header.command;
956        let payload_slice = if full.len() >= 8 { &full[8..] } else { &[] };
957
958        // Connection Validation (cmd=1): respond with CONNECTION_VALIDATED (cmd=9).
959        if cmd_code == 1 {
960            dump_hex_packet(conn_id, "rx", "cmd=1 validation", version, is_be, &full);
961            let validation = spvirit_codec::epics_decode::PvaConnectionValidationPayload::new(
962                payload_slice,
963                is_be,
964                false,
965            );
966            if let Some(val) = validation {
967                debug!(
968                    "Conn {}: validation request (cmd=1) ver={} be={} buf={} qos={} authz={:?}",
969                    conn_id, version, is_be, val.buffer_size, val.qos, val.authz
970                );
971                let resp = spvirit_codec::spvirit_encode::encode_connection_validated(
972                    true, version, is_be,
973                );
974                validate_encoded_packet(conn_id, "conn_validated_resp", &resp);
975                dump_hex_packet(
976                    conn_id,
977                    "tx",
978                    "cmd=9 connection_validated",
979                    version,
980                    is_be,
981                    &resp,
982                );
983                state.registry.send_msg(conn_id, resp).await;
984                continue;
985            }
986        }
987        if cmd_code == 17 {
988            dump_hex_packet(conn_id, "rx", "cmd=17 get_field", version, is_be, &full);
989        }
990
991        match cmd {
992            PvaPacketCommand::Control(payload) => {
993                debug!("Conn {}: control {}", conn_id, payload);
994                if payload.command == 3 {
995                    let resp = encode_control_message(true, is_be, version, 4, payload.data);
996                    state.registry.send_msg(conn_id, resp).await;
997                }
998                continue;
999            }
1000            PvaPacketCommand::ConnectionValidation(_) => {
1001                debug!("Conn {}: validation request (decoded)", conn_id);
1002            }
1003            PvaPacketCommand::ConnectionValidated(_) => {
1004                debug!("Conn {}: validation confirmed (decoded)", conn_id);
1005            }
1006            PvaPacketCommand::CreateChannel(payload) => {
1007                debug!(
1008                    "Conn {}: create_channel count={}",
1009                    conn_id,
1010                    payload.channels.len()
1011                );
1012                for (cid, pv_name) in payload.channels {
1013                    if has_pv(&state, &pv_name).await {
1014                        let sid = state.sid_counter.fetch_add(1, Ordering::SeqCst);
1015                        conn_state.cid_to_sid.insert(cid, sid);
1016                        conn_state.sid_to_pv.insert(sid, pv_name.clone());
1017                        let resp = encode_create_channel_response(cid, sid, version, is_be);
1018                        state.registry.send_msg(conn_id, resp).await;
1019                        info!(
1020                            "Conn {}: channel '{}' cid={} sid={}",
1021                            conn_id, pv_name, cid, sid
1022                        );
1023                    } else {
1024                        let resp = encode_create_channel_error(cid, "PV not found", version, is_be);
1025                        state.registry.send_msg(conn_id, resp).await;
1026                        info!(
1027                            "Conn {}: channel '{}' not found (cid={})",
1028                            conn_id, pv_name, cid
1029                        );
1030                    }
1031                }
1032            }
1033            PvaPacketCommand::Op(payload) => {
1034                if payload.is_server {
1035                    continue;
1036                }
1037                let sid = payload.sid_or_cid;
1038                let ioid = payload.ioid;
1039                debug!(
1040                    "Conn {}: op cmd={} ioid={} sid={} sub=0x{:02x} body_len={}",
1041                    conn_id,
1042                    payload.command,
1043                    ioid,
1044                    sid,
1045                    payload.subcmd,
1046                    payload.body.len()
1047                );
1048                let Some(pv_name) = conn_state.sid_to_pv.get(&sid).cloned() else {
1049                    state
1050                        .registry
1051                        .send_msg(
1052                            conn_id,
1053                            encode_op_error(
1054                                payload.command,
1055                                payload.subcmd,
1056                                ioid,
1057                                "Unknown SID",
1058                                version,
1059                                is_be,
1060                            ),
1061                        )
1062                        .await;
1063                    continue;
1064                };
1065
1066                let is_init = (payload.subcmd & 0x08) != 0;
1067
1068                match payload.command {
1069                    10 => {
1070                        // GET
1071                        if is_init {
1072                            // Init only needs the type descriptor, not the data.
1073                            // Use get_descriptor first; fall back to snapshot.
1074                            let full_desc = if let Some(desc) =
1075                                state.sources.get_descriptor(&pv_name).await
1076                            {
1077                                desc
1078                            } else if let Some(nt) =
1079                                get_nt_snapshot(&state, &pv_name).await
1080                            {
1081                                nt_payload_desc(&nt)
1082                            } else {
1083                                state
1084                                    .registry
1085                                    .send_msg(
1086                                        conn_id,
1087                                        encode_op_error(
1088                                            payload.command,
1089                                            payload.subcmd,
1090                                            ioid,
1091                                            "PV not found",
1092                                            version,
1093                                            is_be,
1094                                        ),
1095                                    )
1096                                    .await;
1097                                continue;
1098                            };
1099                            let pv_req_fields = decode_pv_request_fields(&payload.body, is_be);
1100                            let desc = match &pv_req_fields {
1101                                Some(fields) => filter_structure_desc(&full_desc, fields),
1102                                None => full_desc,
1103                            };
1104                            conn_state.ioid_to_desc.insert(ioid, desc.clone());
1105                            conn_state.ioid_to_pv.insert(ioid, pv_name.clone());
1106                            let resp = encode_op_init_response_desc(
1107                                payload.command,
1108                                ioid,
1109                                0x08,
1110                                &desc,
1111                                version,
1112                                is_be,
1113                            );
1114                            state.registry.send_msg(conn_id, resp).await;
1115                            info!("Conn {}: get init pv='{}' ioid={}", conn_id, pv_name, ioid);
1116                        } else {
1117                            let Some(nt) = get_nt_snapshot(&state, &pv_name).await else {
1118                                state
1119                                    .registry
1120                                    .send_msg(
1121                                        conn_id,
1122                                        encode_op_error(
1123                                            payload.command,
1124                                            payload.subcmd,
1125                                            ioid,
1126                                            "PV has no data yet",
1127                                            version,
1128                                            is_be,
1129                                        ),
1130                                    )
1131                                    .await;
1132                                continue;
1133                            };
1134                            let resp = if let Some(desc) = conn_state.ioid_to_desc.get(&ioid) {
1135                                encode_op_data_response_filtered(
1136                                    10, ioid, &nt, desc, version, is_be,
1137                                )
1138                            } else {
1139                                encode_op_get_data_response_payload(ioid, &nt, version, is_be)
1140                            };
1141                            state.registry.send_msg(conn_id, resp).await;
1142                            debug!("Conn {}: get data pv='{}' ioid={}", conn_id, pv_name, ioid);
1143                        }
1144                    }
1145                    11 => {
1146                        // PUT
1147                        if is_init {
1148                            // Init only needs the type descriptor, not current data.
1149                            // Use get_descriptor first; fall back to snapshot so PUT
1150                            // can target PVs that do not yet have any data.
1151                            let desc = if let Some(desc) =
1152                                state.sources.get_descriptor(&pv_name).await
1153                            {
1154                                desc
1155                            } else if let Some(nt) = get_nt_snapshot(&state, &pv_name).await {
1156                                nt_payload_desc(&nt)
1157                            } else {
1158                                state
1159                                    .registry
1160                                    .send_msg(
1161                                        conn_id,
1162                                        encode_op_error(
1163                                            payload.command,
1164                                            payload.subcmd,
1165                                            ioid,
1166                                            "PV not found",
1167                                            version,
1168                                            is_be,
1169                                        ),
1170                                    )
1171                                    .await;
1172                                continue;
1173                            };
1174                            if !is_virtual_event_pv(&pv_name)
1175                                && !is_writable_pv(&state, &pv_name).await
1176                            {
1177                                let resp = encode_op_put_status_response(
1178                                    ioid,
1179                                    0x08,
1180                                    "Write access denied",
1181                                    version,
1182                                    is_be,
1183                                );
1184                                state.registry.send_msg(conn_id, resp).await;
1185                                continue;
1186                            }
1187                            conn_state.ioid_to_desc.insert(ioid, desc.clone());
1188                            conn_state.ioid_to_pv.insert(ioid, pv_name.clone());
1189                            let resp = encode_op_init_response_desc(
1190                                payload.command,
1191                                ioid,
1192                                0x08,
1193                                &desc,
1194                                version,
1195                                is_be,
1196                            );
1197                            state.registry.send_msg(conn_id, resp).await;
1198                            info!("Conn {}: put init pv='{}' ioid={}", conn_id, pv_name, ioid);
1199                        } else {
1200                            if (payload.subcmd & 0x40) != 0 {
1201                                if !is_virtual_event_pv(&pv_name)
1202                                    && !is_writable_pv(&state, &pv_name).await
1203                                {
1204                                    let resp = encode_op_put_status_response(
1205                                        ioid,
1206                                        0x40,
1207                                        "Write access denied",
1208                                        version,
1209                                        is_be,
1210                                    );
1211                                    state.registry.send_msg(conn_id, resp).await;
1212                                    continue;
1213                                }
1214                                if let Some(nt) = get_nt_snapshot(&state, &pv_name).await {
1215                                    let resp = encode_op_put_getput_response_payload(
1216                                        ioid, &nt, version, is_be,
1217                                    );
1218                                    state.registry.send_msg(conn_id, resp).await;
1219                                    debug!(
1220                                        "Conn {}: put get-put pv='{}' ioid={}",
1221                                        conn_id, pv_name, ioid
1222                                    );
1223                                } else {
1224                                    state
1225                                        .registry
1226                                        .send_msg(
1227                                            conn_id,
1228                                            encode_op_error(
1229                                                payload.command,
1230                                                payload.subcmd,
1231                                                ioid,
1232                                                "PV not found",
1233                                                version,
1234                                                is_be,
1235                                            ),
1236                                        )
1237                                        .await;
1238                                }
1239                                continue;
1240                            }
1241                            let desc = match conn_state.ioid_to_desc.get(&ioid) {
1242                                Some(d) => d.clone(),
1243                                None => {
1244                                    state
1245                                        .registry
1246                                        .send_msg(
1247                                            conn_id,
1248                                            encode_op_error(
1249                                                payload.command,
1250                                                payload.subcmd,
1251                                                ioid,
1252                                                "PUT without init",
1253                                                version,
1254                                                is_be,
1255                                            ),
1256                                        )
1257                                        .await;
1258                                    continue;
1259                                }
1260                            };
1261                            let decoded = decode_put_body(&payload.body, &desc, is_be);
1262                            if let Some(value) = decoded.as_ref() {
1263                                match state.sources.put(&pv_name, value).await {
1264                                    Ok(changed) => {
1265                                        notify_changed_records(&state, changed).await;
1266                                    }
1267                                    Err(msg) => {
1268                                        let resp = encode_op_put_status_response(
1269                                            ioid,
1270                                            payload.subcmd,
1271                                            &msg,
1272                                            version,
1273                                            is_be,
1274                                        );
1275                                        state.registry.send_msg(conn_id, resp).await;
1276                                        continue;
1277                                    }
1278                                }
1279                            } else {
1280                                debug!(
1281                                    "Conn {}: put decode failed ioid={} body_len={}",
1282                                    conn_id,
1283                                    ioid,
1284                                    payload.body.len()
1285                                );
1286                                let resp = encode_op_put_status_response(
1287                                    ioid,
1288                                    payload.subcmd,
1289                                    "cannot decode PUT body",
1290                                    version,
1291                                    is_be,
1292                                );
1293                                state.registry.send_msg(conn_id, resp).await;
1294                                continue;
1295                            }
1296                            let resp = encode_op_put_response(ioid, payload.subcmd, version, is_be);
1297                            state.registry.send_msg(conn_id, resp).await;
1298                            debug!("Conn {}: put data pv='{}' ioid={}", conn_id, pv_name, ioid);
1299                        }
1300                    }
1301                    12 => {
1302                        // PUT_GET
1303                        if is_init {
1304                            // Init only needs the type descriptor, not current data.
1305                            // Use get_descriptor first; fall back to snapshot so that
1306                            // clients can initiate PUT_GET before any data exists.
1307                            let desc = if let Some(desc) =
1308                                state.sources.get_descriptor(&pv_name).await
1309                            {
1310                                desc
1311                            } else if let Some(nt) = get_nt_snapshot(&state, &pv_name).await {
1312                                nt_payload_desc(&nt)
1313                            } else {
1314                                state
1315                                    .registry
1316                                    .send_msg(
1317                                        conn_id,
1318                                        encode_op_error(
1319                                            payload.command,
1320                                            payload.subcmd,
1321                                            ioid,
1322                                            "PV not found",
1323                                            version,
1324                                            is_be,
1325                                        ),
1326                                    )
1327                                    .await;
1328                                continue;
1329                            };
1330                            if !is_virtual_event_pv(&pv_name)
1331                                && !is_writable_pv(&state, &pv_name).await
1332                            {
1333                                let resp = encode_op_put_get_init_error_response(
1334                                    ioid,
1335                                    "Write access denied",
1336                                    version,
1337                                    is_be,
1338                                );
1339                                state.registry.send_msg(conn_id, resp).await;
1340                                continue;
1341                            }
1342                            conn_state.ioid_to_desc.insert(ioid, desc.clone());
1343                            conn_state.ioid_to_pv.insert(ioid, pv_name.clone());
1344                            let resp =
1345                                encode_op_put_get_init_response(ioid, &desc, &desc, version, is_be);
1346                            state.registry.send_msg(conn_id, resp).await;
1347                            info!(
1348                                "Conn {}: put_get init pv='{}' ioid={}",
1349                                conn_id, pv_name, ioid
1350                            );
1351                        } else {
1352                            let desc = match conn_state.ioid_to_desc.get(&ioid) {
1353                                Some(d) => d.clone(),
1354                                None => {
1355                                    state
1356                                        .registry
1357                                        .send_msg(
1358                                            conn_id,
1359                                            encode_op_error(
1360                                                payload.command,
1361                                                payload.subcmd,
1362                                                ioid,
1363                                                "PUT_GET without init",
1364                                                version,
1365                                                is_be,
1366                                            ),
1367                                        )
1368                                        .await;
1369                                    continue;
1370                                }
1371                            };
1372                            let decoded = decode_put_body(&payload.body, &desc, is_be);
1373                            if let Some(value) = decoded.as_ref() {
1374                                match state.sources.put(&pv_name, value).await {
1375                                    Ok(changed) => {
1376                                        notify_changed_records(&state, changed).await;
1377                                    }
1378                                    Err(msg) => {
1379                                        let resp = encode_op_put_get_data_error_response(
1380                                            ioid, &msg, version, is_be,
1381                                        );
1382                                        state.registry.send_msg(conn_id, resp).await;
1383                                        continue;
1384                                    }
1385                                }
1386                            } else {
1387                                debug!(
1388                                    "Conn {}: put_get decode failed ioid={} body_len={}",
1389                                    conn_id,
1390                                    ioid,
1391                                    payload.body.len()
1392                                );
1393                                let resp = encode_op_put_get_data_error_response(
1394                                    ioid,
1395                                    "cannot decode PUT body",
1396                                    version,
1397                                    is_be,
1398                                );
1399                                state.registry.send_msg(conn_id, resp).await;
1400                                continue;
1401                            }
1402                            if let Some(nt) = get_nt_snapshot(&state, &pv_name).await {
1403                                let resp = encode_op_put_get_data_response_payload(
1404                                    ioid, &nt, version, is_be,
1405                                );
1406                                state.registry.send_msg(conn_id, resp).await;
1407                            } else {
1408                                state
1409                                    .registry
1410                                    .send_msg(
1411                                        conn_id,
1412                                        encode_op_error(
1413                                            payload.command,
1414                                            payload.subcmd,
1415                                            ioid,
1416                                            "PV not found",
1417                                            version,
1418                                            is_be,
1419                                        ),
1420                                    )
1421                                    .await;
1422                            }
1423                            debug!(
1424                                "Conn {}: put_get data pv='{}' ioid={}",
1425                                conn_id, pv_name, ioid
1426                            );
1427                        }
1428                    }
1429                    13 => {
1430                        // MONITOR
1431                        if is_init {
1432                            // Init only needs the type descriptor, not the data.
1433                            // Use get_descriptor first; fall back to snapshot. This
1434                            // lets clients subscribe to PVs before any data has been
1435                            // produced (e.g. NTNDArray before acquire). Real monitor
1436                            // updates are pushed once data arrives. Strict clients
1437                            // like p4p treat a MONITOR init error as fatal, so we
1438                            // must not error out when only the descriptor is known.
1439                            let full_desc = if let Some(desc) =
1440                                state.sources.get_descriptor(&pv_name).await
1441                            {
1442                                desc
1443                            } else if let Some(nt) = get_nt_snapshot(&state, &pv_name).await {
1444                                nt_payload_desc(&nt)
1445                            } else {
1446                                state
1447                                    .registry
1448                                    .send_msg(
1449                                        conn_id,
1450                                        encode_op_error(
1451                                            payload.command,
1452                                            payload.subcmd,
1453                                            ioid,
1454                                            "PV not found",
1455                                            version,
1456                                            is_be,
1457                                        ),
1458                                    )
1459                                    .await;
1460                                continue;
1461                            };
1462                            let pv_req_fields = decode_pv_request_fields(&payload.body, is_be);
1463                            let desc = match &pv_req_fields {
1464                                Some(fields) => filter_structure_desc(&full_desc, fields),
1465                                None => full_desc,
1466                            };
1467                            conn_state.ioid_to_desc.insert(ioid, desc.clone());
1468                            conn_state.ioid_to_pv.insert(ioid, pv_name.clone());
1469                            let pipeline_enabled = (payload.subcmd & 0x80) != 0;
1470                            let mut nfree = 0u32;
1471                            if pipeline_enabled && payload.body.len() >= 4 {
1472                                let start = payload.body.len() - 4;
1473                                nfree = if is_be {
1474                                    u32::from_be_bytes([
1475                                        payload.body[start],
1476                                        payload.body[start + 1],
1477                                        payload.body[start + 2],
1478                                        payload.body[start + 3],
1479                                    ])
1480                                } else {
1481                                    u32::from_le_bytes([
1482                                        payload.body[start],
1483                                        payload.body[start + 1],
1484                                        payload.body[start + 2],
1485                                        payload.body[start + 3],
1486                                    ])
1487                                };
1488                            }
1489                            let resp = encode_op_init_response_desc(
1490                                payload.command,
1491                                ioid,
1492                                0x08,
1493                                &desc,
1494                                version,
1495                                is_be,
1496                            );
1497                            state.registry.send_msg(conn_id, resp).await;
1498                            conn_state.ioid_to_monitor.insert(
1499                                ioid,
1500                                MonitorState {
1501                                    running: false,
1502                                    pipeline_enabled,
1503                                    nfree,
1504                                },
1505                            );
1506                            {
1507                                let mut monitors = state.registry.monitors.lock().await;
1508                                monitors
1509                                    .entry(pv_name.clone())
1510                                    .or_default()
1511                                    .push(MonitorSub {
1512                                        conn_id,
1513                                        ioid,
1514                                        version,
1515                                        is_be,
1516                                        running: false,
1517                                        pipeline_enabled,
1518                                        nfree,
1519                                        filtered_desc: conn_state.ioid_to_desc.get(&ioid).cloned(),
1520                                        last_snapshot: None,
1521                                    });
1522                            }
1523                            info!(
1524                                "Conn {}: monitor init pv='{}' ioid={}",
1525                                conn_id, pv_name, ioid
1526                            );
1527                        } else if (payload.subcmd & 0x10) != 0 {
1528                            // Monitor destroy
1529                            if let Some(nt) = get_nt_snapshot(&state, &pv_name).await {
1530                                let resp = encode_monitor_data_response_payload(
1531                                    ioid, 0x10, &nt, version, is_be,
1532                                );
1533                                state.registry.send_msg(conn_id, resp).await;
1534                            }
1535                            state
1536                                .registry
1537                                .remove_monitor_subscription(conn_id, ioid, &pv_name)
1538                                .await;
1539                            conn_state.ioid_to_monitor.remove(&ioid);
1540                            conn_state.ioid_to_pv.remove(&ioid);
1541                            conn_state.ioid_to_desc.remove(&ioid);
1542                            info!("Conn {}: monitor end ioid={}", conn_id, ioid);
1543                        } else if (payload.subcmd & 0x04) != 0 || (payload.subcmd & 0x80) != 0 {
1544                            // Monitor start/stop/pipeline-ack
1545                            let start = (payload.subcmd & 0x44) == 0x44;
1546                            let stop = (payload.subcmd & 0x44) == 0x04;
1547                            let pipeline_ack = (payload.subcmd & 0x80) != 0;
1548                            let mut nfree = None;
1549                            if pipeline_ack && payload.body.len() >= 4 {
1550                                let v = if is_be {
1551                                    u32::from_be_bytes([
1552                                        payload.body[0],
1553                                        payload.body[1],
1554                                        payload.body[2],
1555                                        payload.body[3],
1556                                    ])
1557                                } else {
1558                                    u32::from_le_bytes([
1559                                        payload.body[0],
1560                                        payload.body[1],
1561                                        payload.body[2],
1562                                        payload.body[3],
1563                                    ])
1564                                };
1565                                nfree = Some(v);
1566                            }
1567                            let running = if start {
1568                                true
1569                            } else if stop {
1570                                false
1571                            } else {
1572                                conn_state
1573                                    .ioid_to_monitor
1574                                    .get(&ioid)
1575                                    .map(|m| m.running)
1576                                    .unwrap_or(true)
1577                            };
1578                            state
1579                                .registry
1580                                .update_monitor_subscription(
1581                                    conn_id,
1582                                    ioid,
1583                                    &pv_name,
1584                                    running,
1585                                    nfree,
1586                                    Some(pipeline_ack),
1587                                )
1588                                .await;
1589                            if let Some(mon) = conn_state.ioid_to_monitor.get_mut(&ioid) {
1590                                mon.running = running;
1591                                if pipeline_ack {
1592                                    mon.pipeline_enabled = true;
1593                                }
1594                                if let Some(v) = nfree {
1595                                    if pipeline_ack {
1596                                        mon.nfree = mon.nfree.saturating_add(v);
1597                                    } else {
1598                                        mon.nfree = v;
1599                                    }
1600                                }
1601                            }
1602                            info!(
1603                                "Conn {}: monitor {} ioid={} ack={} nfree={:?}",
1604                                conn_id,
1605                                if start {
1606                                    "start"
1607                                } else if stop {
1608                                    "stop"
1609                                } else {
1610                                    "ack"
1611                                },
1612                                ioid,
1613                                pipeline_ack,
1614                                nfree
1615                            );
1616                            if start {
1617                                if let Some(nt) = get_nt_snapshot(&state, &pv_name).await {
1618                                    state
1619                                        .registry
1620                                        .send_monitor_update_for(&pv_name, conn_id, ioid, &nt)
1621                                        .await;
1622                                }
1623                            }
1624                        }
1625                    }
1626                    20 => {
1627                        // RPC
1628                        if is_server_rpc_pv(&pv_name) {
1629                            handle_server_rpc(
1630                                &state,
1631                                conn_id,
1632                                ioid,
1633                                payload.subcmd,
1634                                version,
1635                                is_be,
1636                            )
1637                            .await;
1638                        } else {
1639                            let is_init = (payload.subcmd & 0x08) != 0;
1640                            if is_init {
1641                                // RPC INIT — acknowledge with status OK
1642                                let resp = encode_op_status_response(
1643                                    20,
1644                                    ioid,
1645                                    payload.subcmd,
1646                                    version,
1647                                    is_be,
1648                                );
1649                                state.registry.send_msg(conn_id, resp).await;
1650                            } else {
1651                                // RPC EXEC — decode self-describing PVD args and
1652                                // delegate to the source registry
1653                                let decoder = PvdDecoder::new(is_be);
1654                                let args = if !payload.body.is_empty() {
1655                                    decoder
1656                                        .parse_introspection_with_len(&payload.body)
1657                                        .and_then(|(desc, consumed)| {
1658                                            decoder
1659                                                .decode_structure(
1660                                                    &payload.body[consumed..],
1661                                                    &desc,
1662                                                )
1663                                                .map(|(val, _)| val)
1664                                        })
1665                                } else {
1666                                    None
1667                                };
1668                                let empty =
1669                                    spvirit_codec::spvd_decode::DecodedValue::Structure(vec![]);
1670                                let args_ref = args.as_ref().unwrap_or(&empty);
1671                                match state.sources.rpc(&pv_name, args_ref).await {
1672                                    Ok(result) => {
1673                                        let resp =
1674                                            encode_op_rpc_data_response_payload(
1675                                                ioid,
1676                                                payload.subcmd,
1677                                                &result,
1678                                                version,
1679                                                is_be,
1680                                            );
1681                                        state.registry.send_msg(conn_id, resp).await;
1682                                    }
1683                                    Err(msg) => {
1684                                        let resp = encode_op_status_error_response(
1685                                            20,
1686                                            ioid,
1687                                            payload.subcmd,
1688                                            &msg,
1689                                            version,
1690                                            is_be,
1691                                        );
1692                                        state.registry.send_msg(conn_id, resp).await;
1693                                    }
1694                                }
1695                            }
1696                        }
1697                    }
1698                    14 | 16 => {
1699                        state
1700                            .registry
1701                            .send_msg(
1702                                conn_id,
1703                                encode_op_error(
1704                                    payload.command,
1705                                    payload.subcmd,
1706                                    ioid,
1707                                    "Operation not supported",
1708                                    version,
1709                                    is_be,
1710                                ),
1711                            )
1712                            .await;
1713                    }
1714                    _ => {
1715                        state
1716                            .registry
1717                            .send_msg(
1718                                conn_id,
1719                                encode_op_error(
1720                                    payload.command,
1721                                    payload.subcmd,
1722                                    ioid,
1723                                    "Operation not supported",
1724                                    version,
1725                                    is_be,
1726                                ),
1727                            )
1728                            .await;
1729                    }
1730                }
1731            }
1732            PvaPacketCommand::DestroyChannel(payload) => {
1733                let sid = payload.sid;
1734                let cid = payload.cid;
1735                conn_state.cid_to_sid.remove(&cid);
1736                conn_state.sid_to_pv.remove(&sid);
1737                info!(
1738                    "Conn {}: channel destroyed sid={} cid={}",
1739                    conn_id, sid, cid
1740                );
1741            }
1742            PvaPacketCommand::DestroyRequest(payload) => {
1743                let ioid = payload.request_id;
1744                if let Some(pv_name) = conn_state.ioid_to_pv.remove(&ioid) {
1745                    state
1746                        .registry
1747                        .remove_monitor_subscription(conn_id, ioid, &pv_name)
1748                        .await;
1749                    conn_state.ioid_to_desc.remove(&ioid);
1750                    conn_state.ioid_to_monitor.remove(&ioid);
1751                    info!("Conn {}: monitor unsubscribed ioid={}", conn_id, ioid);
1752                }
1753            }
1754            PvaPacketCommand::AuthNZ(_) => {
1755                // Silently accept — pvxs and pvAccessCPP ignore AUTHNZ.
1756                debug!("Conn {}: ignoring AUTHNZ", conn_id);
1757            }
1758            PvaPacketCommand::AclChange(_) => {
1759                let resp =
1760                    encode_message_error("ACL_CHANGE command is not supported", version, is_be);
1761                state.registry.send_msg(conn_id, resp).await;
1762            }
1763            PvaPacketCommand::GetField(payload) => {
1764                handle_get_field_request(&state, &conn_state, conn_id, payload, version, is_be)
1765                    .await;
1766            }
1767            PvaPacketCommand::Echo(payload_bytes) => {
1768                let mut resp =
1769                    encode_header(true, is_be, false, version, 2, payload_bytes.len() as u32);
1770                resp.extend_from_slice(&payload_bytes);
1771                state.registry.send_msg(conn_id, resp).await;
1772            }
1773            PvaPacketCommand::Message(_) => {
1774                let resp = encode_message_error("MESSAGE command is not supported", version, is_be);
1775                state.registry.send_msg(conn_id, resp).await;
1776            }
1777            PvaPacketCommand::MultipleData(_) => {
1778                let resp =
1779                    encode_message_error("MULTIPLE_DATA command is not supported", version, is_be);
1780                state.registry.send_msg(conn_id, resp).await;
1781            }
1782            PvaPacketCommand::CancelRequest(_) => {
1783                let resp =
1784                    encode_message_error("CANCEL_REQUEST command is not supported", version, is_be);
1785                state.registry.send_msg(conn_id, resp).await;
1786            }
1787            PvaPacketCommand::OriginTag(_) => {
1788                let resp =
1789                    encode_message_error("ORIGIN_TAG command is not supported", version, is_be);
1790                state.registry.send_msg(conn_id, resp).await;
1791            }
1792            PvaPacketCommand::Search(payload) => {
1793                debug!(
1794                    "Conn {}: TCP search: pv_count={} mask=0x{:02x}",
1795                    conn_id,
1796                    payload.pv_requests.len(),
1797                    payload.mask
1798                );
1799                let accepts_tcp = payload.protocols.is_empty()
1800                    || payload
1801                        .protocols
1802                        .iter()
1803                        .any(|p| p.eq_ignore_ascii_case("tcp"));
1804                if accepts_tcp {
1805                    let all_names = state.sources.names().await;
1806                    let visible_names = collect_visible_pv_names(
1807                        &all_names,
1808                        state.pvlist_mode,
1809                        state.pvlist_allow_pattern.as_ref(),
1810                        state.pvlist_max,
1811                    );
1812                    let mut cids = Vec::new();
1813                    for (cid, name) in &payload.pv_requests {
1814                        if state.sources.has_pv(name).await
1815                            || is_virtual_event_pv(name)
1816                            || (is_pvlist_virtual_pv(name)
1817                                && state.pvlist_mode == PvListMode::List)
1818                            || (is_server_rpc_pv(name) && state.pvlist_mode != PvListMode::Off)
1819                        {
1820                            cids.push(*cid);
1821                            continue;
1822                        }
1823                        if state.pvlist_mode != PvListMode::Off
1824                            && is_pattern_query(name)
1825                            && visible_names.iter().any(|pv| wildcard_match(name, pv))
1826                        {
1827                            cids.push(*cid);
1828                        }
1829                    }
1830                    let server_discovery_ping = payload.pv_requests.is_empty();
1831                    let found = server_discovery_ping || !cids.is_empty();
1832                    let resp_ip = state.advertise_ip.unwrap_or(state.listen_ip);
1833                    let addr_bytes = if resp_ip.is_unspecified() {
1834                        [0u8; 16]
1835                    } else {
1836                        ip_to_bytes(resp_ip)
1837                    };
1838                    let response = encode_search_response(
1839                        state.guid,
1840                        payload.seq,
1841                        addr_bytes,
1842                        state.tcp_port,
1843                        "tcp",
1844                        found,
1845                        &cids,
1846                        version,
1847                        is_be,
1848                    );
1849                    state.registry.send_msg(conn_id, response).await;
1850                    debug!(
1851                        "Conn {}: TCP search responded found={} matches={}",
1852                        conn_id,
1853                        found,
1854                        cids.len()
1855                    );
1856                } else {
1857                    debug!("Conn {}: TCP search: no compatible protocol", conn_id);
1858                }
1859            }
1860            PvaPacketCommand::SearchResponse(_)
1861            | PvaPacketCommand::Beacon(_) => {
1862                let resp =
1863                    encode_message_error("Unexpected command for server endpoint", version, is_be);
1864                state.registry.send_msg(conn_id, resp).await;
1865            }
1866            PvaPacketCommand::Unknown(payload) => {
1867                let resp = encode_message_error(
1868                    &format!("Unknown command {}", payload.command),
1869                    version,
1870                    is_be,
1871                );
1872                state.registry.send_msg(conn_id, resp).await;
1873            }
1874        }
1875    }
1876
1877    state.registry.cleanup_connection(conn_id).await;
1878    let _ = writer_task.await;
1879    Ok(())
1880}