From fdf22b35321d7c1181915d04d48776709f899a01 Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 12 Nov 2026 09:03:17 +0000 Subject: [PATCH 1/2] Support the Force_Synchronization option E1.31-2016 Section 6.2.6 and 11.1.2: when synchronisation on an address is lost (no sync packet within E131_NETWORK_DATA_LOSS_TIMEOUT), a data packet with Force_Synchronization set may be processed without waiting for a sync packet, while one without it is held until synchronisation resumes. The receiver parsed the bit but never used it, so synchronised data was always held, and SacnSource always sent the bit as 0. The receiver now records when it last got a sync packet on each sync address. A data packet with Force_Synchronization set whose sync address has had no sync packet within the timeout (or none yet, as per 11.1.1) is passed up straight away. Without the option, behaviour is unchanged. SacnSource gains force_synchronization/set_force_synchronization to set the option on the data packets it sends. It defaults to off. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01KzaQVQThQpmqhvLapk99Vs --- src/receive.rs | 127 +++++++++++++++++++++++++++++++++++++++++++- src/source.rs | 34 +++++++++++- tests/ipv4_tests.rs | 3 +- tests/src_tests.rs | 12 +++++ 4 files changed, 172 insertions(+), 4 deletions(-) diff --git a/src/receive.rs b/src/receive.rs index 15f4141e..fede53bc 100644 --- a/src/receive.rs +++ b/src/receive.rs @@ -197,6 +197,10 @@ pub struct SacnReceiver { /// Flag which indicates if an `UniverseTimeout` error should be thrown if it is detected that a source has timed out. announce_timeout: bool, + + /// When the last synchronisation packet was received for each synchronisation address. + /// Used to tell if synchronisation on an address has been lost, as per ANSI E1.31-2016 Section 11.1.2. + last_sync_packets: HashMap, } /// Represents an sACN source/sender on the network that has been discovered by this sACN receiver by receiving universe discovery packets. @@ -306,6 +310,7 @@ impl SacnReceiver { announce_source_discovery: ANNOUNCE_SOURCE_DISCOVERY_DEFAULT, announce_stream_termination: ANNOUNCE_STREAM_TERMINATION_DEFAULT, announce_timeout: ANNOUNCE_TIMEOUT_DEFAULT, + last_sync_packets: HashMap::new(), }; sri.listen_universes(&[E131_DISCOVERY_UNIVERSE])?; @@ -736,7 +741,19 @@ impl SacnReceiver { self.announce_timeout, )?; - if data_pkt.synchronization_address == E131_NO_SYNC_ADDR { + // As per ANSI E1.31-2016 Section 6.2.6 and 11.1.2, if the Force_Synchronization option is set and synchronisation on this + // packet's synchronisation address has been lost (or has not started), the data doesn't have to wait for a synchronisation + // packet. Without the option the data is held until synchronisation resumes. + let process_unsynchronized = data_pkt.synchronization_address != E131_NO_SYNC_ADDR + && data_pkt.force_synchronization + && !self.is_synchronized(data_pkt.synchronization_address); + + if process_unsynchronized { + // Keep listening on the synchronisation address so synchronisation can resume. + self.listen_universes(&[data_pkt.synchronization_address])?; + } + + if data_pkt.synchronization_address == E131_NO_SYNC_ADDR || process_unsynchronized { self.clear_waiting_data(data_pkt.universe); let vals: Vec = data_pkt.data.property_values.into_owned(); @@ -859,6 +876,9 @@ impl SacnReceiver { self.announce_timeout, )?; + self.last_sync_packets + .insert(sync_pkt.synchronization_address, Instant::now()); + let res = self.rtrv_waiting_data(sync_pkt.synchronization_address); if res.is_empty() { Ok(None) @@ -974,6 +994,14 @@ impl SacnReceiver { /// Goes through all the waiting data and removes any which has timed out as a sync-packet for it hasn't been received within the `E131_NETWORK_DATA_LOSS_TIMEOUT` /// period as specified by ANSI E1.31-2018 Section 11.1.2. + /// Returns true if a synchronisation packet has been received on the given synchronisation address within + /// `E131_NETWORK_DATA_LOSS_TIMEOUT`, meaning synchronisation on that address is active as per ANSI E1.31-2016 Section 11.1. + fn is_synchronized(&self, sync_addr: u16) -> bool { + self.last_sync_packets + .get(&sync_addr) + .is_some_and(|t| t.elapsed() < E131_NETWORK_DATA_LOSS_TIMEOUT) + } + fn check_waiting_data_timeouts(&mut self) { self.waiting_data .retain(|_uni, data| data.recv_timestamp.elapsed() < E131_NETWORK_DATA_LOSS_TIMEOUT); @@ -2192,6 +2220,103 @@ mod test { assert_eq!(dmx_rcv.discovered_sources[0].pages[0].universes, universes); } + fn synchronized_data_packet<'a>( + universe: u16, + sync_address: u16, + sequence_number: u8, + force_synchronization: bool, + ) -> DataPacketFramingLayer<'a> { + DataPacketFramingLayer { + synchronization_address: sync_address, + force_synchronization, + ..generate_data_packet_framing_layer_seq_num(universe, sequence_number) + } + } + + /// With `Force_Synchronization` set and no synchronisation packet ever received on the address, data isn't held. + #[test] + fn test_force_synchronization_before_sync_starts() { + const UNIVERSE: u16 = 1; + const SYNC_UNI: u16 = 2; + let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), ACN_SDT_MULTICAST_PORT); + let mut dmx_rcv = SacnReceiver::with_ip(addr, None).unwrap(); + dmx_rcv.listen_universes(&[UNIVERSE]).unwrap(); + let cid = Uuid::new_v4(); + + let res = dmx_rcv + .handle_data_packet(cid, synchronized_data_packet(UNIVERSE, SYNC_UNI, 0, true)) + .unwrap() + .expect("data with Force_Synchronization should be passed up when not synchronised"); + assert_eq!(res[0].universe, UNIVERSE); + assert!(dmx_rcv.waiting_data.is_empty()); + // The receiver still listens on the synchronisation address so synchronisation can start. + assert!(dmx_rcv.is_listening(&SYNC_UNI)); + } + + /// While synchronisation is active, data is held until the synchronisation packet even with `Force_Synchronization` set. + #[test] + fn test_force_synchronization_while_synchronized() { + const UNIVERSE: u16 = 1; + const SYNC_UNI: u16 = 2; + let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), ACN_SDT_MULTICAST_PORT); + let mut dmx_rcv = SacnReceiver::with_ip(addr, None).unwrap(); + dmx_rcv.listen_universes(&[UNIVERSE, SYNC_UNI]).unwrap(); + let cid = Uuid::new_v4(); + + // Synchronisation starts with the first synchronisation packet. + assert!(dmx_rcv + .handle_sync_packet(cid, generate_sync_packet_framing_layer_seq_num(SYNC_UNI, 0)) + .unwrap() + .is_none()); + + assert!(dmx_rcv + .handle_data_packet(cid, synchronized_data_packet(UNIVERSE, SYNC_UNI, 0, true)) + .unwrap() + .is_none()); + + let res = dmx_rcv + .handle_sync_packet(cid, generate_sync_packet_framing_layer_seq_num(SYNC_UNI, 1)) + .unwrap() + .expect("held data should be released by the synchronisation packet"); + assert_eq!(res[0].universe, UNIVERSE); + } + + /// Once synchronisation is lost, `Force_Synchronization` set means data is processed without waiting, and unset means the + /// data is held until synchronisation resumes. + #[test] + fn test_force_synchronization_after_sync_lost() { + const UNIVERSE: u16 = 1; + const SYNC_UNI: u16 = 2; + let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), ACN_SDT_MULTICAST_PORT); + let mut dmx_rcv = SacnReceiver::with_ip(addr, None).unwrap(); + dmx_rcv.listen_universes(&[UNIVERSE, SYNC_UNI]).unwrap(); + let cid = Uuid::new_v4(); + + dmx_rcv + .handle_sync_packet(cid, generate_sync_packet_framing_layer_seq_num(SYNC_UNI, 0)) + .unwrap(); + // No synchronisation packet for longer than the network data loss timeout. + dmx_rcv.last_sync_packets.insert( + SYNC_UNI, + Instant::now() - E131_NETWORK_DATA_LOSS_TIMEOUT - Duration::from_millis(100), + ); + + // Force_Synchronization unset: hold the data. + assert!(dmx_rcv + .handle_data_packet(cid, synchronized_data_packet(UNIVERSE, SYNC_UNI, 0, false)) + .unwrap() + .is_none()); + assert!(dmx_rcv.waiting_data.contains_key(&UNIVERSE)); + + // Force_Synchronization set: process the data now, replacing the held data. + let res = dmx_rcv + .handle_data_packet(cid, synchronized_data_packet(UNIVERSE, SYNC_UNI, 1, true)) + .unwrap() + .expect("data with Force_Synchronization should be passed up once synchronisation is lost"); + assert_eq!(res[0].universe, UNIVERSE); + assert!(dmx_rcv.waiting_data.is_empty()); + } + #[test] fn test_handle_multi_page_discovery_packet() { let addr = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), ACN_SDT_MULTICAST_PORT); diff --git a/src/source.rs b/src/source.rs index f701751b..c59d58c1 100644 --- a/src/source.rs +++ b/src/source.rs @@ -108,6 +108,11 @@ struct SacnSourceInternal { /// upon in an untested environment. preview_data: bool, + /// Flag which is included in sACN data packets that have a synchronisation address. It tells receivers what to do if + /// synchronisation is lost, as per ANSI E1.31-2016 Section 6.2.6 (`Force_Synchronization`). If false receivers hold the + /// data until synchronisation resumes, if true they may process the data without waiting for a synchronisation packet. + force_synchronization: bool, + /// The sequence numbers used for data packets, keeps a reference of the next sequence number to use for each universe. /// Sequence numbers are always in the range [0, 255]. data_sequences: RefCell>, @@ -434,6 +439,32 @@ impl SacnSource { Ok(()) } + /// Returns the value of the `Force_Synchronization` option set on data packets sent by this `SacnSource`. + /// + /// For details see (`set_force_synchronization`)[`set_force_synchronization`]. + /// + /// # Errors + /// `SourceCorrupt`: Returned if the Mutex used to control access to the internal sender is poisoned by a thread encountering + /// a panic while accessing causing the source to be left in a potentially inconsistent state. + pub fn force_synchronization(&self) -> Result { + Ok(unlock_internal(&self.internal)?.force_synchronization) + } + + /// Sets the `Force_Synchronization` option on data packets sent by this `SacnSource`, as per ANSI E1.31-2016 Section 6.2.6. + /// + /// The option tells receivers what to do with synchronised data (data sent with a synchronisation address) if synchronisation is lost, + /// meaning no synchronisation packet has been received on that address within `E131_NETWORK_DATA_LOSS_TIMEOUT`: + /// - false (the default): receivers hold the data until synchronisation resumes. + /// - true: receivers may process the data straight away without waiting for a synchronisation packet. + /// + /// # Errors + /// `SourceCorrupt`: Returned if the Mutex used to control access to the internal sender is poisoned by a thread encountering + /// a panic while accessing causing the source to be left in a potentially inconsistent state. + pub fn set_force_synchronization(&mut self, force_synchronization: bool) -> Result<()> { + unlock_internal_mut(&mut self.internal)?.force_synchronization = force_synchronization; + Ok(()) + } + /// Sets the `is_sending_discovery` flag to the given value. /// /// # Arguments @@ -588,6 +619,7 @@ impl SacnSourceInternal { cid, name: name.to_string(), preview_data: false, + force_synchronization: false, data_sequences: RefCell::new(HashMap::new()), sync_sequences: RefCell::new(HashMap::new()), universes: Vec::new(), @@ -833,7 +865,7 @@ impl SacnSourceInternal { sequence_number: sequence, preview_data: self.preview_data, stream_terminated: false, - force_synchronization: false, + force_synchronization: self.force_synchronization, universe, data: DataPacketDmpLayer { property_values: { diff --git a/tests/ipv4_tests.rs b/tests/ipv4_tests.rs index 7cf5c28f..ff4fe8da 100644 --- a/tests/ipv4_tests.rs +++ b/tests/ipv4_tests.rs @@ -2822,8 +2822,7 @@ fn test_send_recv_multiple_sync_universes(){ /// /// This shows that this timeout mechanism to stop universe synchronisation works. /// -/// Note that this library does not attempt to implement the force_synchronisation bit behaviour and so therefore always stops universe synchronisation if the -/// sync packet is not received within the timeout. +/// The sender leaves the `Force_Synchronization` option unset, so the receiver keeps holding data rather than processing it unsynchronised. /// #[test] #[ignore] diff --git a/tests/src_tests.rs b/tests/src_tests.rs index ee9605f4..9c833b2a 100644 --- a/tests/src_tests.rs +++ b/tests/src_tests.rs @@ -702,3 +702,15 @@ const TEST_DATA_MULTIPLE_UNIVERSE: [u8; 712] = [ 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 0, 100, ]; + +/// The `Force_Synchronization` option defaults to off and can be turned on, as per ANSI E1.31-2016 Section 6.2.6. +#[test] +#[serial] +fn test_set_force_synchronization() { + let mut src = SacnSource::new_v4("Source").unwrap(); + assert!(!src.force_synchronization().unwrap()); + src.set_force_synchronization(true).unwrap(); + assert!(src.force_synchronization().unwrap()); + src.set_force_synchronization(false).unwrap(); + assert!(!src.force_synchronization().unwrap()); +} From 887ac27b9b762de205121c0d48f44c7700f369ae Mon Sep 17 00:00:00 2001 From: Claude Date: Thu, 12 Nov 2026 09:47:37 +0000 Subject: [PATCH 2/2] Close test_set_force_synchronization after merging main The merge resolution dropped the closing brace of the test, which nested the source name tests from main inside it so they no longer ran. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01KzaQVQThQpmqhvLapk99Vs --- tests/src_tests.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/src_tests.rs b/tests/src_tests.rs index 2e5c6b7b..ed19781f 100644 --- a/tests/src_tests.rs +++ b/tests/src_tests.rs @@ -713,6 +713,8 @@ fn test_set_force_synchronization() { assert!(src.force_synchronization().unwrap()); src.set_force_synchronization(false).unwrap(); assert!(!src.force_synchronization().unwrap()); +} + /// A 64 byte source name fills the whole field and leaves no room for the null terminator required by /// ANSI E1.31-2016 Section 6.2.2, so it must be rejected. #[test]