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(state: &ServerState, changed: Vec<(String, NtPayload)>) {
400    for (name, payload) in changed {
401        state.beacon_change.fetch_add(1, Ordering::SeqCst);
402        state.registry.notify_monitors(&name, &payload).await;
403    }
404}
405
406// ---------------------------------------------------------------------------
407// GET_FIELD handler
408// ---------------------------------------------------------------------------
409
410async fn handle_get_field_request(
411    state: &ServerState,
412    conn_state: &ConnState,
413    conn_id: u64,
414    payload: spvirit_codec::epics_decode::PvaGetFieldPayload,
415    version: u8,
416    is_be: bool,
417) {
418    if payload.is_server {
419        let resp = encode_get_field_error(
420            payload.cid,
421            "Unexpected server GET_FIELD payload",
422            version,
423            is_be,
424        );
425        state.registry.send_msg(conn_id, resp).await;
426        return;
427    }
428
429    let request_id = payload.ioid.unwrap_or(payload.cid);
430
431    let sid = payload
432        .sid
433        .or_else(|| conn_state.cid_to_sid.get(&payload.cid).copied())
434        .or_else(|| {
435            conn_state
436                .sid_to_pv
437                .contains_key(&payload.cid)
438                .then_some(payload.cid)
439        })
440        .or_else(|| {
441            (payload.cid == 0 && conn_state.sid_to_pv.len() == 1)
442                .then(|| conn_state.sid_to_pv.keys().copied().next())
443                .flatten()
444        });
445
446    if let Some(sid) = sid {
447        if let Some(pv_name) = conn_state.sid_to_pv.get(&sid) {
448            if let Some(nt) = get_nt_snapshot(state, pv_name).await {
449                let full_desc = nt_payload_desc(&nt);
450                let sub = payload.field_name.as_deref().filter(|s| !s.is_empty());
451                let desc = if let Some(field_path) = sub {
452                    match extract_subfield_desc(&full_desc, field_path) {
453                        Some(sub_desc) => sub_desc,
454                        None => {
455                            let resp = encode_get_field_error(
456                                request_id,
457                                &format!("sub-field '{}' not found", field_path),
458                                version,
459                                is_be,
460                            );
461                            state.registry.send_msg(conn_id, resp).await;
462                            return;
463                        }
464                    }
465                } else {
466                    full_desc
467                };
468                let resp = encode_get_field_response(request_id, &desc, version, is_be);
469                dump_hex_packet(conn_id, "tx", "cmd=17 get_field", version, is_be, &resp);
470                state.registry.send_msg(conn_id, resp).await;
471                debug!(
472                    "Conn {}: get_field cid={} sid={:?} ioid={:?} resolved_sid={} pv='{}' field={:?}",
473                    conn_id,
474                    payload.cid,
475                    payload.sid,
476                    payload.ioid,
477                    sid,
478                    pv_name,
479                    payload.field_name
480                );
481                return;
482            }
483            let resp = encode_get_field_error(request_id, "PV not found", version, is_be);
484            state.registry.send_msg(conn_id, resp).await;
485            return;
486        }
487    }
488
489    if state.pvlist_mode != PvListMode::List {
490        let resp = encode_get_field_error(
491            request_id,
492            "GET_FIELD listing is disabled (set --pvlist-mode=list)",
493            version,
494            is_be,
495        );
496        state.registry.send_msg(conn_id, resp).await;
497        return;
498    }
499
500    let Some(pattern) = requested_pvlist_pattern(payload.field_name.as_deref()) else {
501        let resp = encode_get_field_error(
502            request_id,
503            "GET_FIELD requires a valid list pattern",
504            version,
505            is_be,
506        );
507        state.registry.send_msg(conn_id, resp).await;
508        return;
509    };
510
511    let all_names = state.sources.names().await;
512    let mut names = collect_visible_pv_names(
513        &all_names,
514        state.pvlist_mode,
515        state.pvlist_allow_pattern.as_ref(),
516        state.pvlist_max,
517    );
518    if pattern != "*" {
519        names.retain(|name| wildcard_match(pattern, name));
520    }
521    if names.is_empty() {
522        let resp =
523            encode_get_field_error(request_id, "No PVs matched list request", version, is_be);
524        state.registry.send_msg(conn_id, resp).await;
525        return;
526    }
527    let desc = build_pvlist_structure(&names);
528    let resp = encode_get_field_response(request_id, &desc, version, is_be);
529    dump_hex_packet(
530        conn_id,
531        "tx",
532        "cmd=17 get_field_list",
533        version,
534        is_be,
535        &resp,
536    );
537    state.registry.send_msg(conn_id, resp).await;
538    debug!(
539        "Conn {}: get_field list pattern='{}' returned {} entries",
540        conn_id,
541        pattern,
542        names.len()
543    );
544}
545
546// ---------------------------------------------------------------------------
547// Server RPC handler
548// ---------------------------------------------------------------------------
549
550async fn handle_server_rpc(
551    state: &ServerState,
552    conn_id: u64,
553    ioid: u32,
554    subcmd: u8,
555    version: u8,
556    is_be: bool,
557) {
558    if state.pvlist_mode != PvListMode::List {
559        let resp = encode_op_status_error_response(
560            20,
561            ioid,
562            subcmd,
563            "RPC list endpoint disabled (set --pvlist-mode=list)",
564            version,
565            is_be,
566        );
567        state.registry.send_msg(conn_id, resp).await;
568        return;
569    }
570
571    let all_names = state.sources.names().await;
572    let names = collect_visible_pv_names(
573        &all_names,
574        state.pvlist_mode,
575        state.pvlist_allow_pattern.as_ref(),
576        state.pvlist_max,
577    );
578    let payload = NtPayload::ScalarArray(NtScalarArray::from_value(ScalarArrayValue::Str(names)));
579
580    let is_init = (subcmd & 0x08) != 0;
581    if is_init {
582        let resp = encode_op_status_response(20, ioid, subcmd, version, is_be);
583        state.registry.send_msg(conn_id, resp).await;
584        return;
585    }
586
587    let resp = encode_op_rpc_data_response_payload(ioid, subcmd, &payload, version, is_be);
588    state.registry.send_msg(conn_id, resp).await;
589}
590
591// ---------------------------------------------------------------------------
592// Control message handler (inside segmented stream)
593// ---------------------------------------------------------------------------
594
595async fn handle_control_message(state: &ServerState, conn_id: u64, header: &PvaHeader) {
596    debug!(
597        "Conn {}: control (segmented) cmd={} data={}",
598        conn_id, header.command, header.payload_length
599    );
600    if header.command == 3 {
601        let resp = encode_control_message(
602            true,
603            header.flags.is_msb,
604            header.version,
605            4,
606            header.payload_length,
607        );
608        state.registry.send_msg(conn_id, resp).await;
609    }
610}
611
612// ---------------------------------------------------------------------------
613// UDP search handler
614// ---------------------------------------------------------------------------
615
616/// Run the UDP search responder.
617pub async fn run_udp_search(
618    state: Arc<ServerState>,
619    addr: SocketAddr,
620    tcp_port: u16,
621    guid: [u8; 12],
622    advertise_ip: Option<IpAddr>,
623) -> Result<(), Box<dyn std::error::Error>> {
624    let socket = bind_udp_search_socket(addr)?;
625    socket.set_broadcast(true)?;
626    let mut buf = vec![0u8; 4096];
627
628    loop {
629        let (len, peer) = socket.recv_from(&mut buf).await?;
630        let data = &buf[..len];
631        let header = PvaHeader::new(data);
632        if header.flags.is_control || header.command != 3 {
633            continue;
634        }
635        let mut pkt = PvaPacket::new(data);
636        let Some(cmd) = pkt.decode_payload() else {
637            continue;
638        };
639        let version = pkt.header.version;
640        let is_be = pkt.header.flags.is_msb;
641        match cmd {
642            PvaPacketCommand::Search(payload) => {
643                debug!(
644                    "UDP search from {}: pv_count={} mask=0x{:02x}",
645                    peer,
646                    payload.pv_requests.len(),
647                    payload.mask
648                );
649                let accepts_tcp = payload.protocols.is_empty()
650                    || payload
651                        .protocols
652                        .iter()
653                        .any(|p| p.eq_ignore_ascii_case("tcp"));
654                if !accepts_tcp {
655                    debug!("UDP search: no compatible protocol (tcp not accepted)");
656                    continue;
657                }
658                let all_names = state.sources.names().await;
659                let visible_names = collect_visible_pv_names(
660                    &all_names,
661                    state.pvlist_mode,
662                    state.pvlist_allow_pattern.as_ref(),
663                    state.pvlist_max,
664                );
665                let mut cids = Vec::new();
666                for (cid, name) in &payload.pv_requests {
667                    if state.sources.has_pv(name).await
668                        || is_virtual_event_pv(name)
669                        || (is_pvlist_virtual_pv(name) && state.pvlist_mode == PvListMode::List)
670                        || (is_server_rpc_pv(name) && state.pvlist_mode != PvListMode::Off)
671                    {
672                        cids.push(*cid);
673                        continue;
674                    }
675                    if state.pvlist_mode != PvListMode::Off
676                        && is_pattern_query(name)
677                        && visible_names.iter().any(|pv| wildcard_match(name, pv))
678                    {
679                        cids.push(*cid);
680                    }
681                }
682                let response_required = (payload.mask & 0x01) != 0;
683                let server_discovery_ping = payload.pv_requests.is_empty();
684                let found = server_discovery_ping || !cids.is_empty();
685                if !found && !response_required {
686                    debug!("UDP search: no matches and response not required");
687                    continue;
688                }
689                let resp_ip = if let Some(ip) = advertise_ip {
690                    ip
691                } else if !addr.ip().is_unspecified() {
692                    addr.ip()
693                } else if let Some(ip) = infer_udp_response_ip(peer) {
694                    debug!("UDP search: inferred response address {}", ip);
695                    ip
696                } else {
697                    IpAddr::V4(Ipv4Addr::UNSPECIFIED)
698                };
699                let addr_bytes = if resp_ip.is_unspecified() {
700                    debug!("UDP search: responding with zero address (unspecified listen)");
701                    [0u8; 16]
702                } else {
703                    ip_to_bytes(resp_ip)
704                };
705                let response = encode_search_response(
706                    guid,
707                    payload.seq,
708                    addr_bytes,
709                    tcp_port,
710                    "tcp",
711                    found,
712                    &cids,
713                    version,
714                    is_be,
715                );
716                let reply_target = search_reply_target(&payload.addr, payload.port, peer);
717                if let Err(e) = socket.send_to(&response, reply_target).await {
718                    debug!(
719                        "UDP search: failed sending {} matches to {}: {}",
720                        cids.len(),
721                        reply_target,
722                        e
723                    );
724                    continue;
725                }
726                debug!(
727                    "UDP search: responded found={} with {} matches to {}",
728                    found,
729                    cids.len(),
730                    reply_target
731                );
732            }
733            _ => {}
734        }
735    }
736}
737
738// ---------------------------------------------------------------------------
739// TCP server
740// ---------------------------------------------------------------------------
741
742/// Accept TCP connections and spawn a handler for each.
743pub async fn run_tcp_server(
744    state: Arc<ServerState>,
745    addr: SocketAddr,
746    conn_timeout: Duration,
747) -> Result<(), Box<dyn std::error::Error>> {
748    let listener = TcpListener::bind(addr).await?;
749    let conn_id = Arc::new(std::sync::atomic::AtomicU64::new(1));
750
751    loop {
752        let (stream, peer) = listener.accept().await?;
753        let id = conn_id.fetch_add(1, Ordering::SeqCst);
754        info!("TCP connection {} from {}", id, peer);
755        let state_clone = state.clone();
756        tokio::spawn(async move {
757            if let Err(e) = handle_connection(state_clone, stream, id, conn_timeout).await {
758                error!("Connection {} error: {}", id, e);
759            }
760        });
761    }
762}
763
764// ---------------------------------------------------------------------------
765// Core TCP connection handler
766// ---------------------------------------------------------------------------
767
768/// Handle a single PVA TCP connection.
769///
770/// This is the main protocol loop: handshake, then dispatch each command
771/// (CreateChannel, GET, PUT, PUT_GET, MONITOR, RPC, etc.) using the
772/// [`SourceRegistry`] abstraction.
773pub async fn handle_connection(
774    state: Arc<ServerState>,
775    stream: TcpStream,
776    conn_id: u64,
777    conn_timeout: Duration,
778) -> Result<(), Box<dyn std::error::Error>> {
779    let (mut reader, mut writer) = stream.into_split();
780    let (tx, mut rx) = mpsc::channel::<Vec<u8>>(128);
781
782    {
783        let mut conns = state.registry.conns.lock().await;
784        conns.insert(conn_id, tx);
785    }
786
787    let writer_task = tokio::spawn(async move {
788        while let Some(msg) = rx.recv().await {
789            if writer.write_all(&msg).await.is_err() {
790                break;
791            }
792        }
793    });
794
795    let mut conn_state = ConnState::default();
796
797    // Per EPICS PVA protocol: send SET_BYTE_ORDER control message before validation.
798    let set_byte_order = encode_control_message(true, false, 2, 2, 0);
799    validate_encoded_packet(conn_id, "set_byte_order", &set_byte_order);
800    dump_hex_packet(
801        conn_id,
802        "tx",
803        "ctrl=2 set_byte_order",
804        2,
805        false,
806        &set_byte_order,
807    );
808    state.registry.send_msg(conn_id, set_byte_order).await;
809
810    // Server sends Connection Validation (cmd=1) next.
811    let server_validation =
812        encode_connection_validation(16_384, 512, &["anonymous", "ca"], 2, false);
813    validate_encoded_packet(conn_id, "server_validation_init", &server_validation);
814    dump_hex_packet(
815        conn_id,
816        "tx",
817        "cmd=1 server_validation_init",
818        2,
819        false,
820        &server_validation,
821    );
822    state.registry.send_msg(conn_id, server_validation).await;
823
824    let mut last_activity = Instant::now();
825
826    loop {
827        let mut header = [0u8; 8];
828        let elapsed = last_activity.elapsed();
829        if elapsed >= conn_timeout {
830            info!("Conn {} idle timeout", conn_id);
831            break;
832        }
833        let remaining = conn_timeout - elapsed;
834        let read_header = tokio::time::timeout(remaining, reader.read_exact(&mut header)).await;
835        match read_header {
836            Ok(Ok(_)) => {}
837            Ok(Err(_)) => break,
838            Err(_) => {
839                info!("Conn {} idle timeout", conn_id);
840                break;
841            }
842        }
843        let header_pkt = PvaPacket::new(&header);
844        let payload_len = if header_pkt.header.flags.is_control {
845            0usize
846        } else {
847            header_pkt.header.payload_length as usize
848        };
849        let mut payload = vec![0u8; payload_len];
850        if payload_len > 0 {
851            let elapsed = last_activity.elapsed();
852            if elapsed >= conn_timeout {
853                info!("Conn {} idle timeout", conn_id);
854                break;
855            }
856            let remaining = conn_timeout - elapsed;
857            let read_payload =
858                tokio::time::timeout(remaining, reader.read_exact(&mut payload)).await;
859            match read_payload {
860                Ok(Ok(_)) => {}
861                Ok(Err(_)) => break,
862                Err(_) => {
863                    info!("Conn {} idle timeout", conn_id);
864                    break;
865                }
866            }
867        }
868        last_activity = Instant::now();
869        let mut full = header.to_vec();
870        full.extend_from_slice(&payload);
871
872        // Segmented message reassembly
873        if header_pkt.header.flags.is_segmented != 0 && !header_pkt.header.flags.is_control {
874            debug!(
875                "Conn {}: segmented message cmd={} seg=0x{:02x}",
876                conn_id, header_pkt.header.command, header_pkt.header.flags.is_segmented
877            );
878            let mut payloads = vec![payload];
879            let mut seg_flags = header_pkt.header.flags;
880            while !seg_flags.is_last_segment {
881                let mut seg_header = [0u8; 8];
882                let elapsed = last_activity.elapsed();
883                if elapsed >= conn_timeout {
884                    info!("Conn {} idle timeout", conn_id);
885                    break;
886                }
887                let remaining = conn_timeout - elapsed;
888                let read_header =
889                    tokio::time::timeout(remaining, reader.read_exact(&mut seg_header)).await;
890                match read_header {
891                    Ok(Ok(_)) => {}
892                    Ok(Err(_)) => break,
893                    Err(_) => {
894                        info!("Conn {} idle timeout", conn_id);
895                        break;
896                    }
897                }
898
899                let seg_header_pkt = PvaPacket::new(&seg_header);
900                let seg_payload_len = if seg_header_pkt.header.flags.is_control {
901                    0usize
902                } else {
903                    seg_header_pkt.header.payload_length as usize
904                };
905                let mut seg_payload = vec![0u8; seg_payload_len];
906                if seg_payload_len > 0 {
907                    let elapsed = last_activity.elapsed();
908                    if elapsed >= conn_timeout {
909                        info!("Conn {} idle timeout", conn_id);
910                        break;
911                    }
912                    let remaining = conn_timeout - elapsed;
913                    let read_payload =
914                        tokio::time::timeout(remaining, reader.read_exact(&mut seg_payload)).await;
915                    match read_payload {
916                        Ok(Ok(_)) => {}
917                        Ok(Err(_)) => break,
918                        Err(_) => {
919                            info!("Conn {} idle timeout", conn_id);
920                            break;
921                        }
922                    }
923                }
924                last_activity = Instant::now();
925
926                if seg_header_pkt.header.flags.is_control {
927                    handle_control_message(&state, conn_id, &seg_header_pkt.header).await;
928                    continue;
929                }
930                if seg_header_pkt.header.flags.is_segmented == 0 {
931                    debug!(
932                        "Conn {}: segmented message interrupted by non-segmented cmd={}",
933                        conn_id, seg_header_pkt.header.command
934                    );
935                    break;
936                }
937                payloads.push(seg_payload);
938                seg_flags = seg_header_pkt.header.flags;
939            }
940            full = assemble_segmented_message(header, payloads);
941        }
942
943        let mut pkt = PvaPacket::new(&full);
944        let Some(cmd) = pkt.decode_payload() else {
945            continue;
946        };
947        let version = pkt.header.version;
948        let is_be = pkt.header.flags.is_msb;
949        let cmd_code = pkt.header.command;
950        let payload_slice = if full.len() >= 8 { &full[8..] } else { &[] };
951
952        // Connection Validation (cmd=1): respond with CONNECTION_VALIDATED (cmd=9).
953        if cmd_code == 1 {
954            dump_hex_packet(conn_id, "rx", "cmd=1 validation", version, is_be, &full);
955            let validation = spvirit_codec::epics_decode::PvaConnectionValidationPayload::new(
956                payload_slice,
957                is_be,
958                false,
959            );
960            if let Some(val) = validation {
961                debug!(
962                    "Conn {}: validation request (cmd=1) ver={} be={} buf={} qos={} authz={:?}",
963                    conn_id, version, is_be, val.buffer_size, val.qos, val.authz
964                );
965                let resp = spvirit_codec::spvirit_encode::encode_connection_validated(
966                    true, version, is_be,
967                );
968                validate_encoded_packet(conn_id, "conn_validated_resp", &resp);
969                dump_hex_packet(
970                    conn_id,
971                    "tx",
972                    "cmd=9 connection_validated",
973                    version,
974                    is_be,
975                    &resp,
976                );
977                state.registry.send_msg(conn_id, resp).await;
978                continue;
979            }
980        }
981        if cmd_code == 17 {
982            dump_hex_packet(conn_id, "rx", "cmd=17 get_field", version, is_be, &full);
983        }
984
985        match cmd {
986            PvaPacketCommand::Control(payload) => {
987                debug!("Conn {}: control {}", conn_id, payload);
988                if payload.command == 3 {
989                    let resp = encode_control_message(true, is_be, version, 4, payload.data);
990                    state.registry.send_msg(conn_id, resp).await;
991                }
992                continue;
993            }
994            PvaPacketCommand::ConnectionValidation(_) => {
995                debug!("Conn {}: validation request (decoded)", conn_id);
996            }
997            PvaPacketCommand::ConnectionValidated(_) => {
998                debug!("Conn {}: validation confirmed (decoded)", conn_id);
999            }
1000            PvaPacketCommand::CreateChannel(payload) => {
1001                debug!(
1002                    "Conn {}: create_channel count={}",
1003                    conn_id,
1004                    payload.channels.len()
1005                );
1006                for (cid, pv_name) in payload.channels {
1007                    if has_pv(&state, &pv_name).await {
1008                        let sid = state.sid_counter.fetch_add(1, Ordering::SeqCst);
1009                        conn_state.cid_to_sid.insert(cid, sid);
1010                        conn_state.sid_to_pv.insert(sid, pv_name.clone());
1011                        let resp = encode_create_channel_response(cid, sid, version, is_be);
1012                        state.registry.send_msg(conn_id, resp).await;
1013                        info!(
1014                            "Conn {}: channel '{}' cid={} sid={}",
1015                            conn_id, pv_name, cid, sid
1016                        );
1017                    } else {
1018                        let resp = encode_create_channel_error(cid, "PV not found", version, is_be);
1019                        state.registry.send_msg(conn_id, resp).await;
1020                        info!(
1021                            "Conn {}: channel '{}' not found (cid={})",
1022                            conn_id, pv_name, cid
1023                        );
1024                    }
1025                }
1026            }
1027            PvaPacketCommand::Op(payload) => {
1028                if payload.is_server {
1029                    continue;
1030                }
1031                let sid = payload.sid_or_cid;
1032                let ioid = payload.ioid;
1033                debug!(
1034                    "Conn {}: op cmd={} ioid={} sid={} sub=0x{:02x} body_len={}",
1035                    conn_id,
1036                    payload.command,
1037                    ioid,
1038                    sid,
1039                    payload.subcmd,
1040                    payload.body.len()
1041                );
1042                let Some(pv_name) = conn_state.sid_to_pv.get(&sid).cloned() else {
1043                    state
1044                        .registry
1045                        .send_msg(
1046                            conn_id,
1047                            encode_op_error(
1048                                payload.command,
1049                                payload.subcmd,
1050                                ioid,
1051                                "Unknown SID",
1052                                version,
1053                                is_be,
1054                            ),
1055                        )
1056                        .await;
1057                    continue;
1058                };
1059
1060                let is_init = (payload.subcmd & 0x08) != 0;
1061
1062                match payload.command {
1063                    10 => {
1064                        // GET
1065                        if is_init {
1066                            // Init only needs the type descriptor, not the data.
1067                            // Use get_descriptor first; fall back to snapshot.
1068                            let full_desc =
1069                                if let Some(desc) = state.sources.get_descriptor(&pv_name).await {
1070                                    desc
1071                                } else if let Some(nt) = get_nt_snapshot(&state, &pv_name).await {
1072                                    nt_payload_desc(&nt)
1073                                } else {
1074                                    state
1075                                        .registry
1076                                        .send_msg(
1077                                            conn_id,
1078                                            encode_op_error(
1079                                                payload.command,
1080                                                payload.subcmd,
1081                                                ioid,
1082                                                "PV not found",
1083                                                version,
1084                                                is_be,
1085                                            ),
1086                                        )
1087                                        .await;
1088                                    continue;
1089                                };
1090                            let pv_req_fields = decode_pv_request_fields(&payload.body, is_be);
1091                            let desc = match &pv_req_fields {
1092                                Some(fields) => filter_structure_desc(&full_desc, fields),
1093                                None => full_desc,
1094                            };
1095                            conn_state.ioid_to_desc.insert(ioid, desc.clone());
1096                            conn_state.ioid_to_pv.insert(ioid, pv_name.clone());
1097                            let resp = encode_op_init_response_desc(
1098                                payload.command,
1099                                ioid,
1100                                0x08,
1101                                &desc,
1102                                version,
1103                                is_be,
1104                            );
1105                            state.registry.send_msg(conn_id, resp).await;
1106                            info!("Conn {}: get init pv='{}' ioid={}", conn_id, pv_name, ioid);
1107                        } else {
1108                            let Some(nt) = get_nt_snapshot(&state, &pv_name).await else {
1109                                state
1110                                    .registry
1111                                    .send_msg(
1112                                        conn_id,
1113                                        encode_op_error(
1114                                            payload.command,
1115                                            payload.subcmd,
1116                                            ioid,
1117                                            "PV has no data yet",
1118                                            version,
1119                                            is_be,
1120                                        ),
1121                                    )
1122                                    .await;
1123                                continue;
1124                            };
1125                            let resp = if let Some(desc) = conn_state.ioid_to_desc.get(&ioid) {
1126                                encode_op_data_response_filtered(
1127                                    10, ioid, &nt, desc, version, is_be,
1128                                )
1129                            } else {
1130                                encode_op_get_data_response_payload(ioid, &nt, version, is_be)
1131                            };
1132                            state.registry.send_msg(conn_id, resp).await;
1133                            debug!("Conn {}: get data pv='{}' ioid={}", conn_id, pv_name, ioid);
1134                        }
1135                    }
1136                    11 => {
1137                        // PUT
1138                        if is_init {
1139                            // Init only needs the type descriptor, not current data.
1140                            // Use get_descriptor first; fall back to snapshot so PUT
1141                            // can target PVs that do not yet have any data.
1142                            let desc =
1143                                if let Some(desc) = state.sources.get_descriptor(&pv_name).await {
1144                                    desc
1145                                } else if let Some(nt) = get_nt_snapshot(&state, &pv_name).await {
1146                                    nt_payload_desc(&nt)
1147                                } else {
1148                                    state
1149                                        .registry
1150                                        .send_msg(
1151                                            conn_id,
1152                                            encode_op_error(
1153                                                payload.command,
1154                                                payload.subcmd,
1155                                                ioid,
1156                                                "PV not found",
1157                                                version,
1158                                                is_be,
1159                                            ),
1160                                        )
1161                                        .await;
1162                                    continue;
1163                                };
1164                            if !is_virtual_event_pv(&pv_name)
1165                                && !is_writable_pv(&state, &pv_name).await
1166                            {
1167                                let resp = encode_op_put_status_response(
1168                                    ioid,
1169                                    0x08,
1170                                    "Write access denied",
1171                                    version,
1172                                    is_be,
1173                                );
1174                                state.registry.send_msg(conn_id, resp).await;
1175                                continue;
1176                            }
1177                            conn_state.ioid_to_desc.insert(ioid, desc.clone());
1178                            conn_state.ioid_to_pv.insert(ioid, pv_name.clone());
1179                            let resp = encode_op_init_response_desc(
1180                                payload.command,
1181                                ioid,
1182                                0x08,
1183                                &desc,
1184                                version,
1185                                is_be,
1186                            );
1187                            state.registry.send_msg(conn_id, resp).await;
1188                            info!("Conn {}: put init pv='{}' ioid={}", conn_id, pv_name, ioid);
1189                        } else {
1190                            if (payload.subcmd & 0x40) != 0 {
1191                                if !is_virtual_event_pv(&pv_name)
1192                                    && !is_writable_pv(&state, &pv_name).await
1193                                {
1194                                    let resp = encode_op_put_status_response(
1195                                        ioid,
1196                                        0x40,
1197                                        "Write access denied",
1198                                        version,
1199                                        is_be,
1200                                    );
1201                                    state.registry.send_msg(conn_id, resp).await;
1202                                    continue;
1203                                }
1204                                if let Some(nt) = get_nt_snapshot(&state, &pv_name).await {
1205                                    let resp = encode_op_put_getput_response_payload(
1206                                        ioid, &nt, version, is_be,
1207                                    );
1208                                    state.registry.send_msg(conn_id, resp).await;
1209                                    debug!(
1210                                        "Conn {}: put get-put pv='{}' ioid={}",
1211                                        conn_id, pv_name, ioid
1212                                    );
1213                                } else {
1214                                    state
1215                                        .registry
1216                                        .send_msg(
1217                                            conn_id,
1218                                            encode_op_error(
1219                                                payload.command,
1220                                                payload.subcmd,
1221                                                ioid,
1222                                                "PV not found",
1223                                                version,
1224                                                is_be,
1225                                            ),
1226                                        )
1227                                        .await;
1228                                }
1229                                continue;
1230                            }
1231                            let desc = match conn_state.ioid_to_desc.get(&ioid) {
1232                                Some(d) => d.clone(),
1233                                None => {
1234                                    state
1235                                        .registry
1236                                        .send_msg(
1237                                            conn_id,
1238                                            encode_op_error(
1239                                                payload.command,
1240                                                payload.subcmd,
1241                                                ioid,
1242                                                "PUT without init",
1243                                                version,
1244                                                is_be,
1245                                            ),
1246                                        )
1247                                        .await;
1248                                    continue;
1249                                }
1250                            };
1251                            let decoded = decode_put_body(&payload.body, &desc, is_be);
1252                            if let Some(value) = decoded.as_ref() {
1253                                match state.sources.put(&pv_name, value).await {
1254                                    Ok(changed) => {
1255                                        notify_changed_records(&state, changed).await;
1256                                    }
1257                                    Err(msg) => {
1258                                        let resp = encode_op_put_status_response(
1259                                            ioid,
1260                                            payload.subcmd,
1261                                            &msg,
1262                                            version,
1263                                            is_be,
1264                                        );
1265                                        state.registry.send_msg(conn_id, resp).await;
1266                                        continue;
1267                                    }
1268                                }
1269                            } else {
1270                                debug!(
1271                                    "Conn {}: put decode failed ioid={} body_len={}",
1272                                    conn_id,
1273                                    ioid,
1274                                    payload.body.len()
1275                                );
1276                                let resp = encode_op_put_status_response(
1277                                    ioid,
1278                                    payload.subcmd,
1279                                    "cannot decode PUT body",
1280                                    version,
1281                                    is_be,
1282                                );
1283                                state.registry.send_msg(conn_id, resp).await;
1284                                continue;
1285                            }
1286                            let resp = encode_op_put_response(ioid, payload.subcmd, version, is_be);
1287                            state.registry.send_msg(conn_id, resp).await;
1288                            debug!("Conn {}: put data pv='{}' ioid={}", conn_id, pv_name, ioid);
1289                        }
1290                    }
1291                    12 => {
1292                        // PUT_GET
1293                        if is_init {
1294                            // Init only needs the type descriptor, not current data.
1295                            // Use get_descriptor first; fall back to snapshot so that
1296                            // clients can initiate PUT_GET before any data exists.
1297                            let desc =
1298                                if let Some(desc) = state.sources.get_descriptor(&pv_name).await {
1299                                    desc
1300                                } else if let Some(nt) = get_nt_snapshot(&state, &pv_name).await {
1301                                    nt_payload_desc(&nt)
1302                                } else {
1303                                    state
1304                                        .registry
1305                                        .send_msg(
1306                                            conn_id,
1307                                            encode_op_error(
1308                                                payload.command,
1309                                                payload.subcmd,
1310                                                ioid,
1311                                                "PV not found",
1312                                                version,
1313                                                is_be,
1314                                            ),
1315                                        )
1316                                        .await;
1317                                    continue;
1318                                };
1319                            if !is_virtual_event_pv(&pv_name)
1320                                && !is_writable_pv(&state, &pv_name).await
1321                            {
1322                                let resp = encode_op_put_get_init_error_response(
1323                                    ioid,
1324                                    "Write access denied",
1325                                    version,
1326                                    is_be,
1327                                );
1328                                state.registry.send_msg(conn_id, resp).await;
1329                                continue;
1330                            }
1331                            conn_state.ioid_to_desc.insert(ioid, desc.clone());
1332                            conn_state.ioid_to_pv.insert(ioid, pv_name.clone());
1333                            let resp =
1334                                encode_op_put_get_init_response(ioid, &desc, &desc, version, is_be);
1335                            state.registry.send_msg(conn_id, resp).await;
1336                            info!(
1337                                "Conn {}: put_get init pv='{}' ioid={}",
1338                                conn_id, pv_name, ioid
1339                            );
1340                        } else {
1341                            let desc = match conn_state.ioid_to_desc.get(&ioid) {
1342                                Some(d) => d.clone(),
1343                                None => {
1344                                    state
1345                                        .registry
1346                                        .send_msg(
1347                                            conn_id,
1348                                            encode_op_error(
1349                                                payload.command,
1350                                                payload.subcmd,
1351                                                ioid,
1352                                                "PUT_GET without init",
1353                                                version,
1354                                                is_be,
1355                                            ),
1356                                        )
1357                                        .await;
1358                                    continue;
1359                                }
1360                            };
1361                            let decoded = decode_put_body(&payload.body, &desc, is_be);
1362                            if let Some(value) = decoded.as_ref() {
1363                                match state.sources.put(&pv_name, value).await {
1364                                    Ok(changed) => {
1365                                        notify_changed_records(&state, changed).await;
1366                                    }
1367                                    Err(msg) => {
1368                                        let resp = encode_op_put_get_data_error_response(
1369                                            ioid, &msg, version, is_be,
1370                                        );
1371                                        state.registry.send_msg(conn_id, resp).await;
1372                                        continue;
1373                                    }
1374                                }
1375                            } else {
1376                                debug!(
1377                                    "Conn {}: put_get decode failed ioid={} body_len={}",
1378                                    conn_id,
1379                                    ioid,
1380                                    payload.body.len()
1381                                );
1382                                let resp = encode_op_put_get_data_error_response(
1383                                    ioid,
1384                                    "cannot decode PUT body",
1385                                    version,
1386                                    is_be,
1387                                );
1388                                state.registry.send_msg(conn_id, resp).await;
1389                                continue;
1390                            }
1391                            if let Some(nt) = get_nt_snapshot(&state, &pv_name).await {
1392                                let resp = encode_op_put_get_data_response_payload(
1393                                    ioid, &nt, version, is_be,
1394                                );
1395                                state.registry.send_msg(conn_id, resp).await;
1396                            } else {
1397                                state
1398                                    .registry
1399                                    .send_msg(
1400                                        conn_id,
1401                                        encode_op_error(
1402                                            payload.command,
1403                                            payload.subcmd,
1404                                            ioid,
1405                                            "PV not found",
1406                                            version,
1407                                            is_be,
1408                                        ),
1409                                    )
1410                                    .await;
1411                            }
1412                            debug!(
1413                                "Conn {}: put_get data pv='{}' ioid={}",
1414                                conn_id, pv_name, ioid
1415                            );
1416                        }
1417                    }
1418                    13 => {
1419                        // MONITOR
1420                        if is_init {
1421                            // Init only needs the type descriptor, not the data.
1422                            // Use get_descriptor first; fall back to snapshot. This
1423                            // lets clients subscribe to PVs before any data has been
1424                            // produced (e.g. NTNDArray before acquire). Real monitor
1425                            // updates are pushed once data arrives. Strict clients
1426                            // like p4p treat a MONITOR init error as fatal, so we
1427                            // must not error out when only the descriptor is known.
1428                            let full_desc =
1429                                if let Some(desc) = state.sources.get_descriptor(&pv_name).await {
1430                                    desc
1431                                } else if let Some(nt) = get_nt_snapshot(&state, &pv_name).await {
1432                                    nt_payload_desc(&nt)
1433                                } else {
1434                                    state
1435                                        .registry
1436                                        .send_msg(
1437                                            conn_id,
1438                                            encode_op_error(
1439                                                payload.command,
1440                                                payload.subcmd,
1441                                                ioid,
1442                                                "PV not found",
1443                                                version,
1444                                                is_be,
1445                                            ),
1446                                        )
1447                                        .await;
1448                                    continue;
1449                                };
1450                            let pv_req_fields = decode_pv_request_fields(&payload.body, is_be);
1451                            let desc = match &pv_req_fields {
1452                                Some(fields) => filter_structure_desc(&full_desc, fields),
1453                                None => full_desc,
1454                            };
1455                            conn_state.ioid_to_desc.insert(ioid, desc.clone());
1456                            conn_state.ioid_to_pv.insert(ioid, pv_name.clone());
1457                            let pipeline_enabled = (payload.subcmd & 0x80) != 0;
1458                            let mut nfree = 0u32;
1459                            if pipeline_enabled && payload.body.len() >= 4 {
1460                                let start = payload.body.len() - 4;
1461                                nfree = if is_be {
1462                                    u32::from_be_bytes([
1463                                        payload.body[start],
1464                                        payload.body[start + 1],
1465                                        payload.body[start + 2],
1466                                        payload.body[start + 3],
1467                                    ])
1468                                } else {
1469                                    u32::from_le_bytes([
1470                                        payload.body[start],
1471                                        payload.body[start + 1],
1472                                        payload.body[start + 2],
1473                                        payload.body[start + 3],
1474                                    ])
1475                                };
1476                            }
1477                            let resp = encode_op_init_response_desc(
1478                                payload.command,
1479                                ioid,
1480                                0x08,
1481                                &desc,
1482                                version,
1483                                is_be,
1484                            );
1485                            state.registry.send_msg(conn_id, resp).await;
1486                            conn_state.ioid_to_monitor.insert(
1487                                ioid,
1488                                MonitorState {
1489                                    running: false,
1490                                    pipeline_enabled,
1491                                    nfree,
1492                                },
1493                            );
1494                            {
1495                                let mut monitors = state.registry.monitors.lock().await;
1496                                monitors
1497                                    .entry(pv_name.clone())
1498                                    .or_default()
1499                                    .push(MonitorSub {
1500                                        conn_id,
1501                                        ioid,
1502                                        version,
1503                                        is_be,
1504                                        running: false,
1505                                        pipeline_enabled,
1506                                        nfree,
1507                                        filtered_desc: conn_state.ioid_to_desc.get(&ioid).cloned(),
1508                                        last_snapshot: None,
1509                                    });
1510                            }
1511                            info!(
1512                                "Conn {}: monitor init pv='{}' ioid={}",
1513                                conn_id, pv_name, ioid
1514                            );
1515                        } else if (payload.subcmd & 0x10) != 0 {
1516                            // Monitor destroy
1517                            if let Some(nt) = get_nt_snapshot(&state, &pv_name).await {
1518                                let resp = encode_monitor_data_response_payload(
1519                                    ioid, 0x10, &nt, version, is_be,
1520                                );
1521                                state.registry.send_msg(conn_id, resp).await;
1522                            }
1523                            state
1524                                .registry
1525                                .remove_monitor_subscription(conn_id, ioid, &pv_name)
1526                                .await;
1527                            conn_state.ioid_to_monitor.remove(&ioid);
1528                            conn_state.ioid_to_pv.remove(&ioid);
1529                            conn_state.ioid_to_desc.remove(&ioid);
1530                            info!("Conn {}: monitor end ioid={}", conn_id, ioid);
1531                        } else if (payload.subcmd & 0x04) != 0 || (payload.subcmd & 0x80) != 0 {
1532                            // Monitor start/stop/pipeline-ack
1533                            let start = (payload.subcmd & 0x44) == 0x44;
1534                            let stop = (payload.subcmd & 0x44) == 0x04;
1535                            let pipeline_ack = (payload.subcmd & 0x80) != 0;
1536                            let mut nfree = None;
1537                            if pipeline_ack && payload.body.len() >= 4 {
1538                                let v = if is_be {
1539                                    u32::from_be_bytes([
1540                                        payload.body[0],
1541                                        payload.body[1],
1542                                        payload.body[2],
1543                                        payload.body[3],
1544                                    ])
1545                                } else {
1546                                    u32::from_le_bytes([
1547                                        payload.body[0],
1548                                        payload.body[1],
1549                                        payload.body[2],
1550                                        payload.body[3],
1551                                    ])
1552                                };
1553                                nfree = Some(v);
1554                            }
1555                            let running = if start {
1556                                true
1557                            } else if stop {
1558                                false
1559                            } else {
1560                                conn_state
1561                                    .ioid_to_monitor
1562                                    .get(&ioid)
1563                                    .map(|m| m.running)
1564                                    .unwrap_or(true)
1565                            };
1566                            state
1567                                .registry
1568                                .update_monitor_subscription(
1569                                    conn_id,
1570                                    ioid,
1571                                    &pv_name,
1572                                    running,
1573                                    nfree,
1574                                    Some(pipeline_ack),
1575                                )
1576                                .await;
1577                            if let Some(mon) = conn_state.ioid_to_monitor.get_mut(&ioid) {
1578                                mon.running = running;
1579                                if pipeline_ack {
1580                                    mon.pipeline_enabled = true;
1581                                }
1582                                if let Some(v) = nfree {
1583                                    if pipeline_ack {
1584                                        mon.nfree = mon.nfree.saturating_add(v);
1585                                    } else {
1586                                        mon.nfree = v;
1587                                    }
1588                                }
1589                            }
1590                            info!(
1591                                "Conn {}: monitor {} ioid={} ack={} nfree={:?}",
1592                                conn_id,
1593                                if start {
1594                                    "start"
1595                                } else if stop {
1596                                    "stop"
1597                                } else {
1598                                    "ack"
1599                                },
1600                                ioid,
1601                                pipeline_ack,
1602                                nfree
1603                            );
1604                            if start {
1605                                if let Some(nt) = get_nt_snapshot(&state, &pv_name).await {
1606                                    state
1607                                        .registry
1608                                        .send_monitor_update_for(&pv_name, conn_id, ioid, &nt)
1609                                        .await;
1610                                }
1611                            }
1612                        }
1613                    }
1614                    20 => {
1615                        // RPC
1616                        if is_server_rpc_pv(&pv_name) {
1617                            handle_server_rpc(
1618                                &state,
1619                                conn_id,
1620                                ioid,
1621                                payload.subcmd,
1622                                version,
1623                                is_be,
1624                            )
1625                            .await;
1626                        } else {
1627                            let is_init = (payload.subcmd & 0x08) != 0;
1628                            if is_init {
1629                                // RPC INIT — acknowledge with status OK
1630                                let resp = encode_op_status_response(
1631                                    20,
1632                                    ioid,
1633                                    payload.subcmd,
1634                                    version,
1635                                    is_be,
1636                                );
1637                                state.registry.send_msg(conn_id, resp).await;
1638                            } else {
1639                                // RPC EXEC — decode self-describing PVD args and
1640                                // delegate to the source registry
1641                                let decoder = PvdDecoder::new(is_be);
1642                                let args = if !payload.body.is_empty() {
1643                                    decoder
1644                                        .parse_introspection_with_len(&payload.body)
1645                                        .and_then(|(desc, consumed)| {
1646                                            decoder
1647                                                .decode_structure(&payload.body[consumed..], &desc)
1648                                                .map(|(val, _)| val)
1649                                        })
1650                                } else {
1651                                    None
1652                                };
1653                                let empty =
1654                                    spvirit_codec::spvd_decode::DecodedValue::Structure(vec![]);
1655                                let args_ref = args.as_ref().unwrap_or(&empty);
1656                                match state.sources.rpc(&pv_name, args_ref).await {
1657                                    Ok(result) => {
1658                                        let resp = encode_op_rpc_data_response_payload(
1659                                            ioid,
1660                                            payload.subcmd,
1661                                            &result,
1662                                            version,
1663                                            is_be,
1664                                        );
1665                                        state.registry.send_msg(conn_id, resp).await;
1666                                    }
1667                                    Err(msg) => {
1668                                        let resp = encode_op_status_error_response(
1669                                            20,
1670                                            ioid,
1671                                            payload.subcmd,
1672                                            &msg,
1673                                            version,
1674                                            is_be,
1675                                        );
1676                                        state.registry.send_msg(conn_id, resp).await;
1677                                    }
1678                                }
1679                            }
1680                        }
1681                    }
1682                    14 | 16 => {
1683                        state
1684                            .registry
1685                            .send_msg(
1686                                conn_id,
1687                                encode_op_error(
1688                                    payload.command,
1689                                    payload.subcmd,
1690                                    ioid,
1691                                    "Operation not supported",
1692                                    version,
1693                                    is_be,
1694                                ),
1695                            )
1696                            .await;
1697                    }
1698                    _ => {
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            }
1716            PvaPacketCommand::DestroyChannel(payload) => {
1717                let sid = payload.sid;
1718                let cid = payload.cid;
1719                conn_state.cid_to_sid.remove(&cid);
1720                conn_state.sid_to_pv.remove(&sid);
1721                info!(
1722                    "Conn {}: channel destroyed sid={} cid={}",
1723                    conn_id, sid, cid
1724                );
1725            }
1726            PvaPacketCommand::DestroyRequest(payload) => {
1727                let ioid = payload.request_id;
1728                if let Some(pv_name) = conn_state.ioid_to_pv.remove(&ioid) {
1729                    state
1730                        .registry
1731                        .remove_monitor_subscription(conn_id, ioid, &pv_name)
1732                        .await;
1733                    conn_state.ioid_to_desc.remove(&ioid);
1734                    conn_state.ioid_to_monitor.remove(&ioid);
1735                    info!("Conn {}: monitor unsubscribed ioid={}", conn_id, ioid);
1736                }
1737            }
1738            PvaPacketCommand::AuthNZ(_) => {
1739                // Silently accept — pvxs and pvAccessCPP ignore AUTHNZ.
1740                debug!("Conn {}: ignoring AUTHNZ", conn_id);
1741            }
1742            PvaPacketCommand::AclChange(_) => {
1743                let resp =
1744                    encode_message_error("ACL_CHANGE command is not supported", version, is_be);
1745                state.registry.send_msg(conn_id, resp).await;
1746            }
1747            PvaPacketCommand::GetField(payload) => {
1748                handle_get_field_request(&state, &conn_state, conn_id, payload, version, is_be)
1749                    .await;
1750            }
1751            PvaPacketCommand::Echo(payload_bytes) => {
1752                let mut resp =
1753                    encode_header(true, is_be, false, version, 2, payload_bytes.len() as u32);
1754                resp.extend_from_slice(&payload_bytes);
1755                state.registry.send_msg(conn_id, resp).await;
1756            }
1757            PvaPacketCommand::Message(_) => {
1758                let resp = encode_message_error("MESSAGE command is not supported", version, is_be);
1759                state.registry.send_msg(conn_id, resp).await;
1760            }
1761            PvaPacketCommand::MultipleData(_) => {
1762                let resp =
1763                    encode_message_error("MULTIPLE_DATA command is not supported", version, is_be);
1764                state.registry.send_msg(conn_id, resp).await;
1765            }
1766            PvaPacketCommand::CancelRequest(_) => {
1767                let resp =
1768                    encode_message_error("CANCEL_REQUEST command is not supported", version, is_be);
1769                state.registry.send_msg(conn_id, resp).await;
1770            }
1771            PvaPacketCommand::OriginTag(_) => {
1772                let resp =
1773                    encode_message_error("ORIGIN_TAG command is not supported", version, is_be);
1774                state.registry.send_msg(conn_id, resp).await;
1775            }
1776            PvaPacketCommand::Search(payload) => {
1777                debug!(
1778                    "Conn {}: TCP search: pv_count={} mask=0x{:02x}",
1779                    conn_id,
1780                    payload.pv_requests.len(),
1781                    payload.mask
1782                );
1783                let accepts_tcp = payload.protocols.is_empty()
1784                    || payload
1785                        .protocols
1786                        .iter()
1787                        .any(|p| p.eq_ignore_ascii_case("tcp"));
1788                if accepts_tcp {
1789                    let all_names = state.sources.names().await;
1790                    let visible_names = collect_visible_pv_names(
1791                        &all_names,
1792                        state.pvlist_mode,
1793                        state.pvlist_allow_pattern.as_ref(),
1794                        state.pvlist_max,
1795                    );
1796                    let mut cids = Vec::new();
1797                    for (cid, name) in &payload.pv_requests {
1798                        if state.sources.has_pv(name).await
1799                            || is_virtual_event_pv(name)
1800                            || (is_pvlist_virtual_pv(name) && state.pvlist_mode == PvListMode::List)
1801                            || (is_server_rpc_pv(name) && state.pvlist_mode != PvListMode::Off)
1802                        {
1803                            cids.push(*cid);
1804                            continue;
1805                        }
1806                        if state.pvlist_mode != PvListMode::Off
1807                            && is_pattern_query(name)
1808                            && visible_names.iter().any(|pv| wildcard_match(name, pv))
1809                        {
1810                            cids.push(*cid);
1811                        }
1812                    }
1813                    let server_discovery_ping = payload.pv_requests.is_empty();
1814                    let found = server_discovery_ping || !cids.is_empty();
1815                    let resp_ip = state.advertise_ip.unwrap_or(state.listen_ip);
1816                    let addr_bytes = if resp_ip.is_unspecified() {
1817                        [0u8; 16]
1818                    } else {
1819                        ip_to_bytes(resp_ip)
1820                    };
1821                    let response = encode_search_response(
1822                        state.guid,
1823                        payload.seq,
1824                        addr_bytes,
1825                        state.tcp_port,
1826                        "tcp",
1827                        found,
1828                        &cids,
1829                        version,
1830                        is_be,
1831                    );
1832                    state.registry.send_msg(conn_id, response).await;
1833                    debug!(
1834                        "Conn {}: TCP search responded found={} matches={}",
1835                        conn_id,
1836                        found,
1837                        cids.len()
1838                    );
1839                } else {
1840                    debug!("Conn {}: TCP search: no compatible protocol", conn_id);
1841                }
1842            }
1843            PvaPacketCommand::SearchResponse(_) | PvaPacketCommand::Beacon(_) => {
1844                let resp =
1845                    encode_message_error("Unexpected command for server endpoint", version, is_be);
1846                state.registry.send_msg(conn_id, resp).await;
1847            }
1848            PvaPacketCommand::Unknown(payload) => {
1849                let resp = encode_message_error(
1850                    &format!("Unknown command {}", payload.command),
1851                    version,
1852                    is_be,
1853                );
1854                state.registry.send_msg(conn_id, resp).await;
1855            }
1856        }
1857    }
1858
1859    state.registry.cleanup_connection(conn_id).await;
1860    let _ = writer_task.await;
1861    Ok(())
1862}