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