Reset outstation event buffer on session teardown (#423)

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);
             }
 

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;
 

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;
 

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,

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

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();
 }

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;

dnp3/src/outstation/tests/harness/harness.rs → dnp3/src/outstation/tests/harness/runner.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<Connected>`, so using them on a disconnected harness is a compile error
+/// rather than a runtime panic.
+pub(crate) type OutstationHarness = Harness<Connected>;
+
+type Session = JoinHandle<(OutstationTask, RunError)>;
+
+pub(crate) struct Harness<S> {
     pub(crate) handle: OutstationHandle,
-    pub(crate) io: sfio_tokio_mock_io::Handle,
-    task: JoinHandle<RunError>,
+    state: S,
     events: EventReceiver,
     pub(crate) application_data: Arc<Mutex<ApplicationData>>,
 }
 
-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<u8> {
-        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<S> Harness<S> {
     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<Connected> {
+    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<u8> {
+        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<Disconnected> {
+        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<Disconnected> {
+    /// 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,
     }