From 00efc3bf0fef2e82f67ccecb854f34f33b8229d2 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Tue, 16 Jun 2026 12:25:38 -0700 Subject: [PATCH 1/4] reset outstation event buffer on session teardown (#423) The outstation reused the same Session and DatabaseHandle across reconnects without re-arming the event buffer's per-event selection state. An event transmitted but not confirmed before a link drop (or TCP-server session replacement) was stranded in the Written state: never re-selected, never re-reported, and not even advertised in the class IIN. Reset per-connection state (event buffer, transport reader/writer, and session state) from a single RAII guard in OutstationTask::run so it runs on both normal return and future cancellation. This makes session.run() cancel-safe, which the TCP server relies on when it drops the running session future to accept a replacement connection. Add a reconnect-capable, type-state outstation test harness and a regression test that drops the link before CONFIRM and asserts the unconfirmed event is re-armed and re-reported on the next connection. Closes #423. --- dnp3/src/outstation/session.rs | 8 +- dnp3/src/outstation/task.rs | 34 +++-- dnp3/src/outstation/tests/freeze.rs | 2 +- dnp3/src/outstation/tests/harness/harness.rs | 145 ++++++++++++++----- dnp3/src/outstation/tests/read_states.rs | 35 +++++ dnp3/src/tcp/outstation/server_task.rs | 2 + 6 files changed, 177 insertions(+), 49 deletions(-) diff --git a/dnp3/src/outstation/session.rs b/dnp3/src/outstation/session.rs index 672c23ae0..a85321683 100644 --- a/dnp3/src/outstation/session.rs +++ b/dnp3/src/outstation/session.rs @@ -374,12 +374,18 @@ impl OutstationSession { ) -> RunError { loop { if let Err(err) = self.run_idle_state(io, reader, writer, database).await { - self.state.reset(); + // teardown (including self.state.reset()) is handled by SessionTeardown + // in OutstationTask::run, so it also covers a cancelled run future return err; } } } + /// reset the per-connection session state between communication sessions + pub(crate) fn reset_session_state(&mut self) { + self.state.reset(); + } + async fn write_unsolicited( &mut self, io: &mut PhysLayer, diff --git a/dnp3/src/outstation/task.rs b/dnp3/src/outstation/task.rs index 04f264790..6c5a570de 100644 --- a/dnp3/src/outstation/task.rs +++ b/dnp3/src/outstation/task.rs @@ -35,6 +35,21 @@ pub(crate) struct OutstationTask { database: DatabaseHandle, } +/// RAII guard that re-arms per-connection state on session teardown. +struct SessionTeardown<'a> { + task: &'a mut OutstationTask, +} + +impl Drop for SessionTeardown<'_> { + fn drop(&mut self) { + // re-arm, don't clear: unconfirmed events return to Unselected for re-delivery + self.task.database.reset(); + self.task.reader.reset(); + self.task.writer.reset(); + self.task.session.reset_session_state(); + } +} + impl OutstationTask { /// create an `OutstationTask` and return it along with a `DatabaseHandle` for updating it #[allow(clippy::too_many_arguments)] @@ -97,17 +112,16 @@ impl OutstationTask { self.session.change_master_address(address); } - /// run the outstation task asynchronously until a `SessionError` occurs + /// run the outstation task asynchronously until a `RunError` occurs + /// + /// Cancel-safe: per-connection state is re-armed by `SessionTeardown`'s `Drop`, so + /// dropping this future tears down exactly as returning from it does. pub(crate) async fn run(&mut self, io: &mut PhysLayer) -> RunError { - let res = self - .session - .run(io, &mut self.reader, &mut self.writer, &mut self.database) - .await; - - self.reader.reset(); - self.writer.reset(); - - res + let guard = SessionTeardown { task: self }; + let task = &mut *guard.task; // reborrow so Drop can reset after the run future releases its borrows + task.session + .run(io, &mut task.reader, &mut task.writer, &mut task.database) + .await } /// process received outstation messages while idle without a session diff --git a/dnp3/src/outstation/tests/freeze.rs b/dnp3/src/outstation/tests/freeze.rs index 7d2058504..dfbcca06d 100644 --- a/dnp3/src/outstation/tests/freeze.rs +++ b/dnp3/src/outstation/tests/freeze.rs @@ -184,5 +184,5 @@ async fn freeze_at_time_no_ack() { )), )]); - assert_eq!(harness.io.pop_event(), None); + harness.expect_no_response(); } diff --git a/dnp3/src/outstation/tests/harness/harness.rs b/dnp3/src/outstation/tests/harness/harness.rs index c6ec0d155..6f36aa657 100644 --- a/dnp3/src/outstation/tests/harness/harness.rs +++ b/dnp3/src/outstation/tests/harness/harness.rs @@ -34,43 +34,48 @@ pub(crate) fn get_default_unsolicited_config() -> OutstationConfig { config } -pub(crate) struct OutstationHarness { +/// Type-state harness: `S` is the connection state and carries the per-state data, so +/// the connected/disconnected shapes are distinct types. IO methods exist only on +/// `Harness`, so using them on a disconnected harness is a compile error +/// rather than a runtime panic. +pub(crate) type OutstationHarness = Harness; + +type Session = JoinHandle<(OutstationTask, RunError)>; + +pub(crate) struct Harness { pub(crate) handle: OutstationHandle, - pub(crate) io: sfio_tokio_mock_io::Handle, - task: JoinHandle, + state: S, events: EventReceiver, pub(crate) application_data: Arc>, } -impl OutstationHarness { - pub(crate) async fn test_request_response(&mut self, request: &[u8], response: &[u8]) { - self.send_and_process(request).await; - self.expect_response(response).await; - } - - pub(crate) async fn expect_write(&mut self) -> Vec { - match self.io.next_event().await { - sfio_tokio_mock_io::Event::Write(bytes) => bytes, - x => panic!("Expected write but got: {x:?}"), - } - } +/// connected state: a live mock IO handle and the running session +pub(crate) struct Connected { + io: sfio_tokio_mock_io::Handle, + session: Session, +} - pub(crate) async fn expect_response(&mut self, response: &[u8]) { - assert_eq!( - self.io.next_event().await, - sfio_tokio_mock_io::Event::Write(response.to_vec()) - ); - } +/// disconnected state: the task (and its `DatabaseHandle`) parked between connections +pub(crate) struct Disconnected { + task: OutstationTask, +} - pub(crate) fn expect_no_response(&mut self) { - assert_eq!(self.io.pop_event(), None); - } +async fn run_session(mut task: OutstationTask, mut io: PhysLayer) -> (OutstationTask, RunError) { + let err = task.run(&mut io).await; + (task, err) +} - pub(crate) async fn send_and_process(&mut self, request: &[u8]) { - self.io.read(request); - assert_eq!(self.io.next_event().await, sfio_tokio_mock_io::Event::Read); +fn spawn_session(task: OutstationTask) -> Connected { + let (io, io_handle) = sfio_tokio_mock_io::mock(); + let phys = PhysLayer::Mock(io); + Connected { + io: io_handle, + session: tokio::spawn(run_session(task, phys)), } +} +// methods available in any connection state +impl Harness { pub(crate) async fn wait_for_events(&mut self, expected: &[Event]) { for event in expected { let next = self.events.next().await; @@ -100,6 +105,79 @@ impl OutstationHarness { } } +// methods that require a live connection +impl Harness { + pub(crate) async fn test_request_response(&mut self, request: &[u8], response: &[u8]) { + self.send_and_process(request).await; + self.expect_response(response).await; + } + + pub(crate) async fn expect_write(&mut self) -> Vec { + match self.state.io.next_event().await { + sfio_tokio_mock_io::Event::Write(bytes) => bytes, + x => panic!("Expected write but got: {x:?}"), + } + } + + pub(crate) async fn expect_response(&mut self, response: &[u8]) { + assert_eq!( + self.state.io.next_event().await, + sfio_tokio_mock_io::Event::Write(response.to_vec()) + ); + } + + pub(crate) fn expect_no_response(&mut self) { + assert_eq!(self.state.io.pop_event(), None); + } + + pub(crate) async fn send_and_process(&mut self, request: &[u8]) { + self.state.io.read(request); + assert_eq!( + self.state.io.next_event().await, + sfio_tokio_mock_io::Event::Read + ); + } + + /// Drop the link via an injected read error and recover the task, asserting the + /// session terminated with a `RunError::Link`. The task (and its `DatabaseHandle`) + /// survive into the returned disconnected harness. + pub(crate) async fn drop_link(self) -> Harness { + let Harness { + handle, + state: Connected { mut io, session }, + events, + application_data, + } = self; + + io.read_error(std::io::ErrorKind::ConnectionReset); + let (task, err) = session.await.expect("outstation task panicked"); + assert!( + matches!(err, RunError::Link(_)), + "expected RunError::Link but got {err:?}" + ); + + Harness { + handle, + state: Disconnected { task }, + events, + application_data, + } + } +} + +impl Harness { + /// Start a new connection reusing the same task + database (as the real + /// TCP/serial transports do across reconnects) + pub(crate) fn reconnect(self) -> OutstationHarness { + Harness { + handle: self.handle, + state: spawn_session(self.state.task), + events: self.events, + application_data: self.application_data, + } + } +} + pub(crate) fn new_harness(config: OutstationConfig) -> OutstationHarness { new_harness_impl(config, None, None) } @@ -127,7 +205,7 @@ fn new_harness_impl( let (data, application) = MockOutstationApplication::create(sender.clone()); - let (task, handle) = OutstationTask::create( + let (mut task, handle) = OutstationTask::create( Enabled::Yes, LinkModes::test(), ParseOptions::get_static(), @@ -138,8 +216,6 @@ fn new_harness_impl( MockControlHandler::create(sender.clone()), ); - let mut task = Box::new(task); - let master_address = master_address.unwrap_or(config.master_address); task.get_reader() @@ -151,14 +227,9 @@ fn new_harness_impl( PhysAddr::None, )); - let (io, io_handle) = sfio_tokio_mock_io::mock(); - - let mut io = PhysLayer::Mock(io); - - OutstationHarness { + Harness { handle, - io: io_handle, - task: tokio::spawn(async move { task.run(&mut io).await }), + state: spawn_session(task), events: receiver, application_data: data, } diff --git a/dnp3/src/outstation/tests/read_states.rs b/dnp3/src/outstation/tests/read_states.rs index 9d84c5073..6fdc44b5e 100644 --- a/dnp3/src/outstation/tests/read_states.rs +++ b/dnp3/src/outstation/tests/read_states.rs @@ -204,3 +204,38 @@ async fn confirm_can_time_out() { .wait_for_events(&[Event::SolicitedConfirmTimeout(0)]) .await; } + +// Regression test for #423: an event transmitted but not confirmed before a hard link +// drop must be RETAINED and RE-ARMED on the next connection against the same database — +// re-advertised in the class IIN and re-reported on read, and NOT leaked into an +// unrelated response. Without the teardown reset the event is stranded in the `Written` +// state: never re-selected, never re-reported, and not even advertised as pending. +#[tokio::test] +async fn re_arms_unconfirmed_events_after_link_drop() { + let mut harness = new_harness(get_default_config()); + + harness.handle.database.transaction(create_binary_and_event); + + // read the event, then drop the link before CONFIRM so it stays in the `Written` state + harness + .test_request_response(READ_CLASS_123, BINARY_EVENT_RESPONSE) + .await; + harness.check_events(&[Event::EnterSolicitedConfirmWait(0)]); + + let harness = harness.drop_link().await; + + // reconnect against the same database + let mut harness = harness.reconnect(); + + // the unconfirmed event is re-armed: advertised as pending in the class IIN, but NOT + // dumped into this unrelated (empty) read + harness + .test_request_response(EMPTY_READ, EMPTY_RESPONSE_WITH_PENDING_EVENTS) + .await; + + // ...and re-reported when its class is read + harness + .test_request_response(READ_CLASS_123, BINARY_EVENT_RESPONSE) + .await; + harness.check_events(&[Event::EnterSolicitedConfirmWait(0)]); +} diff --git a/dnp3/src/tcp/outstation/server_task.rs b/dnp3/src/tcp/outstation/server_task.rs index b2d5544d6..e01b9a175 100644 --- a/dnp3/src/tcp/outstation/server_task.rs +++ b/dnp3/src/tcp/outstation/server_task.rs @@ -63,6 +63,8 @@ impl ServerTask { res = self.session.run(io) => { Err(res) } + // a new connection cancels the session.run() future above; this relies on + // session.run() being cancel-safe x = self.receiver.receive() => { Ok(x?) } From a1bdeb13bd5ad6d170b6f546b65f09b7d3ffd10d Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Tue, 16 Jun 2026 12:28:18 -0700 Subject: [PATCH 2/4] fix build with serialization feature on Rust 1.94 With the serialization feature, serde_json's `impl PartialEq for u32` makes `expected_handle.into()` ambiguous (E0283). Use the explicit `u32::from(..)` conversion. --- dnp3/src/master/tasks/file/close.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/dnp3/src/master/tasks/file/close.rs b/dnp3/src/master/tasks/file/close.rs index 76703f825..28de3390f 100644 --- a/dnp3/src/master/tasks/file/close.rs +++ b/dnp3/src/master/tasks/file/close.rs @@ -56,7 +56,7 @@ impl CloseFileTask { } }; - if obj.file_handle != expected_handle.into() { + if obj.file_handle != u32::from(expected_handle) { return Err(FileError::WrongHandle); } From 9cafa535dd6dd7c0ad2cf21dde860eada2928c1b Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Tue, 16 Jun 2026 12:31:02 -0700 Subject: [PATCH 3/4] fix clippy::field_reassign_with_default in master tests Construct AssociationConfig with struct-update syntax instead of mutating fields after Default::default(). --- dnp3/src/master/tests/auto_tasks.rs | 30 +++++++++++++++++++---------- dnp3/src/master/tests/startup.rs | 24 +++++++++++++++-------- 2 files changed, 36 insertions(+), 18 deletions(-) diff --git a/dnp3/src/master/tests/auto_tasks.rs b/dnp3/src/master/tests/auto_tasks.rs index 0d364188a..dddcb16b1 100644 --- a/dnp3/src/master/tests/auto_tasks.rs +++ b/dnp3/src/master/tests/auto_tasks.rs @@ -12,8 +12,10 @@ const CLASS_1_EVENTS: Iin = Iin::new(Iin1::new(0x02), Iin2::new(0x00)); #[tokio::test] async fn auto_integrity_scan_on_buffer_overflow() { - let mut config = AssociationConfig::default(); - config.auto_integrity_scan_on_buffer_overflow = true; + let config = AssociationConfig { + auto_integrity_scan_on_buffer_overflow: true, + ..Default::default() + }; let mut seq = Sequence::default(); let mut harness = create_association(config).await; @@ -37,8 +39,10 @@ async fn auto_integrity_scan_on_buffer_overflow() { #[tokio::test] async fn auto_integrity_scan_on_buffer_overflow_disabled() { - let mut config = AssociationConfig::default(); - config.auto_integrity_scan_on_buffer_overflow = false; + let config = AssociationConfig { + auto_integrity_scan_on_buffer_overflow: false, + ..Default::default() + }; let mut seq = Sequence::default(); let mut harness = create_association(config).await; @@ -57,8 +61,10 @@ async fn auto_integrity_scan_on_buffer_overflow_disabled() { #[tokio::test] async fn auto_event_class_scan() { - let mut config = AssociationConfig::default(); - config.event_scan_on_events_available = EventClasses::all(); + let config = AssociationConfig { + event_scan_on_events_available: EventClasses::all(), + ..Default::default() + }; let mut seq = Sequence::default(); let mut harness = create_association(config).await; @@ -82,8 +88,10 @@ async fn auto_event_class_scan() { #[tokio::test] async fn auto_event_class_ignore_one_class_scan() { - let mut config = AssociationConfig::default(); - config.event_scan_on_events_available = EventClasses::new(false, true, true); + let config = AssociationConfig { + event_scan_on_events_available: EventClasses::new(false, true, true), + ..Default::default() + }; let mut seq = Sequence::default(); let mut harness = create_association(config).await; @@ -102,8 +110,10 @@ async fn auto_event_class_ignore_one_class_scan() { #[tokio::test] async fn auto_event_class_scan_disabled() { - let mut config = AssociationConfig::default(); - config.event_scan_on_events_available = EventClasses::none(); + let config = AssociationConfig { + event_scan_on_events_available: EventClasses::none(), + ..Default::default() + }; let mut seq = Sequence::default(); let mut harness = create_association(config).await; diff --git a/dnp3/src/master/tests/startup.rs b/dnp3/src/master/tests/startup.rs index cc4db4f82..e665c74e4 100644 --- a/dnp3/src/master/tests/startup.rs +++ b/dnp3/src/master/tests/startup.rs @@ -154,8 +154,10 @@ async fn master_calls_task_fail_when_auto_tasks_returns_iin2_errors() { #[tokio::test] async fn master_startup_procedure_skips_integrity_poll_if_none() { - let mut config = AssociationConfig::default(); - config.startup_integrity_classes = Classes::none(); + let config = AssociationConfig { + startup_integrity_classes: Classes::none(), + ..Default::default() + }; let mut seq = Sequence::default(); let mut harness = create_association(config).await; @@ -191,9 +193,11 @@ async fn master_startup_procedure_skips_integrity_poll_if_none() { #[tokio::test] async fn master_startup_procedure_skips_disable_unsol_if_none() { - let mut config = AssociationConfig::default(); - config.disable_unsol_classes = EventClasses::none(); - config.enable_unsol_classes = EventClasses::none(); + let config = AssociationConfig { + disable_unsol_classes: EventClasses::none(), + enable_unsol_classes: EventClasses::none(), + ..Default::default() + }; let mut seq = Sequence::default(); let mut harness = create_association(config).await; @@ -386,9 +390,13 @@ async fn ignore_duplicate_unsolicited_response() { #[tokio::test] async fn master_startup_retry_procedure() { - let mut config = AssociationConfig::default(); - config.auto_tasks_retry_strategy = - RetryStrategy::new(Duration::from_secs(1), Duration::from_secs(3)); + let config = AssociationConfig { + auto_tasks_retry_strategy: RetryStrategy::new( + Duration::from_secs(1), + Duration::from_secs(3), + ), + ..Default::default() + }; let mut seq = Sequence::default(); let mut harness = create_association(config).await; From 153bdce662328cb515c7e62836d856fc28478f58 Mon Sep 17 00:00:00 2001 From: jadamcrain Date: Tue, 16 Jun 2026 12:33:54 -0700 Subject: [PATCH 4/4] fix clippy::module_inception in outstation test harness Rename the harness submodule file from harness.rs to runner.rs so it no longer shares the name of its containing `harness` module. --- dnp3/src/outstation/tests/harness/mod.rs | 4 ++-- dnp3/src/outstation/tests/harness/{harness.rs => runner.rs} | 0 2 files changed, 2 insertions(+), 2 deletions(-) rename dnp3/src/outstation/tests/harness/{harness.rs => runner.rs} (100%) diff --git a/dnp3/src/outstation/tests/harness/mod.rs b/dnp3/src/outstation/tests/harness/mod.rs index 8944f20e2..8e6bdade9 100644 --- a/dnp3/src/outstation/tests/harness/mod.rs +++ b/dnp3/src/outstation/tests/harness/mod.rs @@ -1,11 +1,11 @@ pub(crate) use application::*; pub(crate) use control::*; pub(crate) use event::*; -pub(crate) use harness::*; pub(crate) use info::*; +pub(crate) use runner::*; mod application; mod control; mod event; -mod harness; mod info; +mod runner; diff --git a/dnp3/src/outstation/tests/harness/harness.rs b/dnp3/src/outstation/tests/harness/runner.rs similarity index 100% rename from dnp3/src/outstation/tests/harness/harness.rs rename to dnp3/src/outstation/tests/harness/runner.rs