From 82e53e377df62dc50315e48b50454b4c44fe6476 Mon Sep 17 00:00:00 2001 From: Jake Huneau Date: Sun, 25 Oct 2026 03:28:35 +0000 Subject: [PATCH] Write message tracking events in order on one thread Each tracking event was written on a new thread of its own so that processing a message did not wait on the database. DbTrackingModule.persist is synchronized, so when a write was slow the threads behind it queued on the lock and took it in no particular order. Under load a message's events could be written out of order and an earlier state overwrite a later one, which is what issue #517 reports: a message that had received its MDN left recorded with the state of "sending started". BaseMsgTrackingModule now hands each event to a single-thread executor of its own, so the writes stay off the processing thread and happen in the order the events were handled. The thread is a daemon so it never keeps the JVM alive. The tracking module is an ActiveModule, so the shutdown drain added in #625 already calls its awaitIdle with the shared deadline after the pollers have finished their sends. awaitIdle now waits for the events queued so far to be written, so the record of a message sent just before shutting down is not lost. DbTrackingModule.stop stops the writer and start gives a restarted module a new one; an event handled after the writer has stopped is written directly rather than dropped. The maintainer's comment on #517 mentions moving tracking onto a queue in 5.0; this is the minimal form of that and should fit into it. TrackingWriteOrderTest writes 200 events through a tracker whose synchronized persist is slow on the first write. With the previous thread per event all 200 were written but out of order in three runs out of three; it now writes them in order. The other tests cover not blocking the caller, awaitIdle and its timeout, writing directly after stopping, restarting, and shutdown waiting for queued events. --- RELEASE-NOTES.md | 11 + .../msgtracking/BaseMsgTrackingModule.java | 69 ++++++- .../msgtracking/DbTrackingModule.java | 2 + .../msgtracking/TrackingWriteOrderTest.java | 193 ++++++++++++++++++ changes.txt | 11 + 5 files changed, 283 insertions(+), 3 deletions(-) create mode 100644 Server/src/test/java/org/openas2/processor/msgtracking/TrackingWriteOrderTest.java 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 ===========================