diff --git a/RELEASE-NOTES.md b/RELEASE-NOTES.md index 489af3a6..e680ae3f 100644 --- a/RELEASE-NOTES.md +++ b/RELEASE-NOTES.md @@ -9,6 +9,17 @@ The release download file is: OpenAS2Server-4.13.0.zip The zip file contains a PDF document (OpenAS2HowTo.pdf) providing information on installing and using the application. ## NOTE: Testing covers Java 17 to 21. +Version UNRELEASED +=========================== + +This is a bugfix release. +1. Fix message tracking records being left with an out of date state under load (issue #517). Each tracking event was + written to the database on a new thread of its own, and threads do not run in the order they are started, so when + several messages were being processed a message's events could be written out of order and an earlier state such as + "sending started" overwrite a later one such as "MDN received", leaving a delivered message recorded as still being + sent. The events are now written in order on one thread per tracking module, still without processing waiting on + the database, and shutting down waits for the events already queued so the record of what was just sent is not lost. + Version 4.14.0 2026-10-13 =========================== diff --git a/Server/src/main/java/org/openas2/processor/msgtracking/BaseMsgTrackingModule.java b/Server/src/main/java/org/openas2/processor/msgtracking/BaseMsgTrackingModule.java index 90414349..3ae89797 100644 --- a/Server/src/main/java/org/openas2/processor/msgtracking/BaseMsgTrackingModule.java +++ b/Server/src/main/java/org/openas2/processor/msgtracking/BaseMsgTrackingModule.java @@ -15,16 +15,79 @@ import java.util.HashMap; import java.util.Map; +import java.util.concurrent.ExecutionException; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.RejectedExecutionException; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.TimeoutException; public abstract class BaseMsgTrackingModule extends BaseProcessorModule implements TrackingModule { + /* + * Tracking events are written on one thread of the module's own, in the order they were handled. + * Each used to be written on a new thread of its own so that processing a message did not wait on + * the database, but threads do not run in the order they are started, so under load a message's + * events could be written out of order and an early state overwrite a later one: a message that + * had received its MDN could be left recorded as still being sent. One thread keeps the writes off + * the processing thread and in order. + */ + private ExecutorService writer = newWriter(); + + private static ExecutorService newWriter() { + return Executors.newSingleThreadExecutor(r -> { + Thread t = new Thread(r, "msg-tracking-writer"); + // Must not keep the JVM alive once everything else has stopped + t.setDaemon(true); + return t; + }); + } + public void handle(String action, Message msg, Map options) throws OpenAS2Exception { Map fields = buildMap(msg, options); - new Thread(() -> { - persist(msg, fields); - }).start(); + try { + writer.execute(() -> persist(msg, fields)); + } catch (RejectedExecutionException e) { + // The writer has been stopped, so write it here rather than lose it + persist(msg, fields); + } + } + + /** + * Waits for the tracking events already handled to be written, so shutting down does not lose + * the record of what was just sent. + */ + @Override + public boolean awaitIdle(long timeoutMillis) throws InterruptedException { + // The writer is first in first out, so once this runs everything handled before it is written + Future marker; + try { + marker = writer.submit(() -> { }); + } catch (RejectedExecutionException e) { + return true; + } + try { + marker.get(timeoutMillis, TimeUnit.MILLISECONDS); + return true; + } catch (TimeoutException e) { + return false; + } catch (ExecutionException e) { + return true; + } + } + + /** Gives a module started again after being stopped a writer to use. */ + protected void startWriter() { + if (writer.isShutdown()) { + writer = newWriter(); + } + } + /** Stops the writer once the events queued have been written; any handled afterwards are written directly. */ + protected void stopWriter() { + writer.shutdown(); } public void init(Session session, Map options) throws OpenAS2Exception { diff --git a/Server/src/main/java/org/openas2/processor/msgtracking/DbTrackingModule.java b/Server/src/main/java/org/openas2/processor/msgtracking/DbTrackingModule.java index 3cddc515..ad6e3957 100644 --- a/Server/src/main/java/org/openas2/processor/msgtracking/DbTrackingModule.java +++ b/Server/src/main/java/org/openas2/processor/msgtracking/DbTrackingModule.java @@ -395,6 +395,7 @@ public boolean isRunning() { } public void start() throws OpenAS2Exception { + startWriter(); if (useEmbeddedDB) { dbHandler = new EmbeddedDBHandler(); } else { @@ -406,6 +407,7 @@ public void start() throws OpenAS2Exception { } public void stop() { + stopWriter(); dbHandler.stop(); } diff --git a/Server/src/test/java/org/openas2/processor/msgtracking/TrackingWriteOrderTest.java b/Server/src/test/java/org/openas2/processor/msgtracking/TrackingWriteOrderTest.java new file mode 100644 index 00000000..ad5eab91 --- /dev/null +++ b/Server/src/test/java/org/openas2/processor/msgtracking/TrackingWriteOrderTest.java @@ -0,0 +1,193 @@ +package org.openas2.processor.msgtracking; + +import org.junit.jupiter.api.AfterAll; +import org.junit.jupiter.api.BeforeAll; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.TestInstance; +import org.junit.jupiter.api.TestInstance.Lifecycle; +import org.openas2.app.BaseServerSetup; +import org.openas2.message.AS2Message; +import org.openas2.message.Message; + +import java.util.ArrayList; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.Map; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies that tracking events are written in the order they happen (issue #517). + *

+ * Each event used to be written on a new thread of its own, and the writes are synchronized, so when + * a write was slow the events behind it queued on the lock and were written in whatever order the + * threads got it. Under load a message's early "sending started" state could be written after its + * later "MDN received" state, leaving a delivered message recorded as still being sent. They are now + * written in order on one thread of the module's own, and shutting down waits for the ones queued. + */ +@TestInstance(Lifecycle.PER_CLASS) +public class TrackingWriteOrderTest extends BaseServerSetup { + + @BeforeAll + public void setUp() throws Exception { + super.createFileSystemResources(this.getClass().getName()); + super.setup(); + } + + @AfterAll + public void tearDown() throws Exception { + super.tearDown(); + } + + @Test + public void eventsAreWrittenInTheOrderTheyHappened() throws Exception { + // The first write is slow, as a database round trip can be, so the rest queue up behind it + RecordingTracker tracker = new RecordingTracker(200, 0); + List expected = new ArrayList(); + for (int i = 0; i < 200; i++) { + expected.add("state-" + i); + track(tracker, "state-" + i); + } + + // Counted rather than waited for with awaitIdle, so this checks only the order the writes happened in + waitForWrites(tracker, expected.size()); + assertEquals(expected, tracker.written, "events written out of order record a stale final state"); + } + + @Test + public void handlingAnEventDoesNotWaitForItToBeWritten() throws Exception { + RecordingTracker tracker = new RecordingTracker(0, 300); + + long started = System.nanoTime(); + track(tracker, "sent"); + long elapsedMillis = (System.nanoTime() - started) / 1_000_000; + + assertTrue(elapsedMillis < 250, "processing a message must not wait on the database, took " + elapsedMillis + "ms"); + assertTrue(tracker.awaitIdle(5000)); + assertEquals(Collections.singletonList("sent"), tracker.written); + } + + @Test + public void waitingForIdleWaitsForTheEventsAlreadyHandled() throws Exception { + RecordingTracker tracker = new RecordingTracker(0, 50); + for (int i = 0; i < 5; i++) { + track(tracker, "state-" + i); + } + + assertTrue(tracker.awaitIdle(5000)); + assertEquals(5, tracker.written.size(), "every event handled before the wait must be written when it returns"); + } + + @Test + public void waitingForIdleGivesUpAtTheTimeout() throws Exception { + RecordingTracker tracker = new RecordingTracker(0, 300); + for (int i = 0; i < 5; i++) { + track(tracker, "state-" + i); + } + + assertFalse(tracker.awaitIdle(200), "five 300ms writes cannot finish within 200ms"); + assertTrue(tracker.awaitIdle(5000)); + } + + @Test + public void anEventAfterTheWriterIsStoppedIsWrittenRatherThanLost() throws Exception { + RecordingTracker tracker = new RecordingTracker(0, 0); + tracker.stopWriter(); + + track(tracker, "late"); + + assertEquals(Collections.singletonList("late"), tracker.written, "it should have been written directly"); + } + + @Test + public void aRestartedModuleWritesInTheBackgroundAgain() throws Exception { + RecordingTracker tracker = new RecordingTracker(0, 300); + tracker.stopWriter(); + tracker.startWriter(); + + long started = System.nanoTime(); + track(tracker, "after-restart"); + long elapsedMillis = (System.nanoTime() - started) / 1_000_000; + + assertTrue(elapsedMillis < 250, "a restarted module should not be left writing on the caller's thread"); + assertTrue(tracker.awaitIdle(5000)); + } + + @Test + public void shuttingDownWaitsForTheEventsQueued() throws Exception { + RecordingTracker tracker = new RecordingTracker(0, 100); + session.getProcessor().getModules().add(tracker); + try { + for (int i = 0; i < 5; i++) { + track(tracker, "state-" + i); + } + + assertTrue(session.drainWorkInProgress(5000)); + assertEquals(5, tracker.written.size(), "shutting down must not lose the record of what was just processed"); + } finally { + session.getProcessor().getModules().remove(tracker); + } + } + + private void waitForWrites(RecordingTracker tracker, int count) throws InterruptedException { + long deadline = System.currentTimeMillis() + 20000; + while (tracker.written.size() < count && System.currentTimeMillis() < deadline) { + Thread.sleep(20); + } + assertEquals(count, tracker.written.size(), "every event should have been written"); + } + + private void track(RecordingTracker tracker, String state) throws Exception { + Map options = new HashMap(); + options.put("STATE", state); + Message msg = new AS2Message(); + msg.setMessageID(""); + tracker.handle(TrackingModule.DO_TRACK_MSG, msg, options); + } + + /** Records the state of each event in the order it is written. */ + private static class RecordingTracker extends BaseMsgTrackingModule { + private final List written = Collections.synchronizedList(new ArrayList()); + private final long firstWriteMillis; + private final long everyWriteMillis; + + RecordingTracker(long firstWriteMillis, long everyWriteMillis) { + this.firstWriteMillis = firstWriteMillis; + this.everyWriteMillis = everyWriteMillis; + } + + // Synchronized as DbTrackingModule.persist is + @Override + protected synchronized void persist(Message msg, Map map) { + try { + Thread.sleep(written.isEmpty() ? Math.max(firstWriteMillis, everyWriteMillis) : everyWriteMillis); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + written.add(map.get(FIELDS.STATE)); + } + + @Override + public boolean isRunning() { + return true; + } + + @Override + public void start() { + startWriter(); + } + + @Override + public void stop() { + stopWriter(); + } + + @Override + public boolean healthcheck(List failures) { + return true; + } + } +} diff --git a/changes.txt b/changes.txt index fdb885bf..cdbc35e6 100644 --- a/changes.txt +++ b/changes.txt @@ -1,5 +1,16 @@ **IMPORTANT NOTE**: Please review upgrade notes in the RELEASE-NOTES.md if you are upgrading +Version UNRELEASED +=========================== + +This is a bugfix release. +1. Fix message tracking records being left with an out of date state under load (issue #517). Each tracking event was + written to the database on a new thread of its own, and threads do not run in the order they are started, so when + several messages were being processed a message's events could be written out of order and an earlier state such as + "sending started" overwrite a later one such as "MDN received", leaving a delivered message recorded as still being + sent. The events are now written in order on one thread per tracking module, still without processing waiting on + the database, and shutting down waits for the events already queued so the record of what was just sent is not lost. + Version 4.14.0 2026-10-13 ===========================