TRUNK-6516: Follow-up: fix tests

Author: ibacher

api/src/main/java/org/openmrs/event/outbox/OutboxEvent.java

@@ -139,6 +139,25 @@ public void setCompletedListeners(String completedListeners) {
 		this.completedListeners = completedListeners;
 	}
 
+	/**
+	 * Copies the persisted (column) state of the given event onto this instance. The identity and the
+	 * {@code creator}/{@code changedBy} associations are intentionally left untouched: the id never
+	 * changes for the same row, and copying lazy proxies across sessions risks
+	 * {@link org.hibernate.LazyInitializationException}. Add any newly persisted scalar column here.
+	 *
+	 * @param other the event whose state should be copied onto this instance
+	 */
+	public void updateFrom(OutboxEvent other) {
+		this.eventType = other.eventType;
+		this.payload = other.payload;
+		this.dateCreated = other.dateCreated;
+		this.dateChanged = other.dateChanged;
+		this.status = other.status;
+		this.errorCount = other.errorCount;
+		this.errorMessage = other.errorMessage;
+		this.completedListeners = other.completedListeners;
+	}
+
 	// --- Auditable Interface Implementation ---
 
 	@Override

api/src/main/java/org/openmrs/event/outbox/OutboxEventService.java

@@ -128,7 +128,15 @@ public boolean lockEventForProcessing(OutboxEvent outboxEvent) {
 		claimQuery.setParameter("now", new Date());
 
 		int updatedCount = claimQuery.executeUpdate();
-		return updatedCount == 1;
+		if (updatedCount != 1) {
+			return false;
+		}
+
+		// The instance itself isn't updated, so we update it here, but first we need to re-read from the database
+		OutboxEvent claimed = sessionFactory.getCurrentSession().get(OutboxEvent.class, outboxEvent.getId());
+		sessionFactory.getCurrentSession().refresh(claimed);
+		outboxEvent.updateFrom(claimed);
+		return true;
 	}
 
 	@Transactional

api/src/main/resources/hibernate.cfg.xml

@@ -48,6 +48,7 @@
 		<!-- These mappings are required because of references in Obs & Concept -->
 		<mapping class="org.openmrs.ObsReferenceRange"/>
 		<mapping class="org.openmrs.ConceptReferenceRange"/>
+		<mapping class="org.openmrs.event.outbox.OutboxEvent"/>
   </session-factory>
 
 </hibernate-configuration>

api/src/test/java/org/openmrs/event/outbox/OutboxEventIT.java

@@ -12,8 +12,10 @@
 import java.util.ArrayList;
 import java.util.List;
 import java.util.UUID;
+import java.util.stream.Collectors;
 
 import org.hibernate.SessionFactory;
+import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.openmrs.GlobalProperty;
@@ -62,6 +64,11 @@ public void setUp() throws Exception {
 		outboxTaskSchedulerInitializer.schedule();
 	}
 
+	@AfterEach
+	public void tearDown() {
+		outboxTaskSchedulerInitializer.deleteScheduledTasks();
+	}
+
 	@Test
 	public void interceptAndSaveToOutbox_shouldSaveEventWhenListenerExists() throws InterruptedException {
 		// Arrange: A patient event is created, and the registry has a listener.
@@ -104,7 +111,7 @@ public void interceptAndSaveToOutbox_shouldSaveEventWhenListenerExists() throws
 		capturedEvents.remove(afterCommitEvent);
 
 		//Assert the ordering of captured outbox events is correct
-		assertThat(capturedEvents, contains(
+		assertThat(describeState(capturedEvents), capturedEvents, contains(
 		    allOf(hasProperty("method", is("onPatientCreated")),
 		        hasProperty("event", hasProperty("uuid", equalTo(event.getUuid())))),
 		    hasProperty("method", is("onPatientCreatedFailingFailed")),
@@ -171,8 +178,8 @@ public void interceptAndSaveToOutbox_shouldSaveEventWhenNoTransaction() {
 		assertThat(adminService.getGlobalProperty("eventPublished", "false"), equalTo("true"));
 
 		// Assert: The outbox table should have just one event
-		List<OutboxEvent> outboxEvents = (List<OutboxEvent>) sessionFactory.getCurrentSession()
-		        .createQuery("from OutboxEvent").list();
+		List<OutboxEvent> outboxEvents = sessionFactory.getCurrentSession()
+		        .createQuery("from OutboxEvent", OutboxEvent.class).list();
 
 		assertThat(outboxEvents,
 		    contains(
@@ -249,6 +256,24 @@ public Object getNonSerializableObject() {
 		}
 	}
 
+	/**
+	 * Builds a snapshot of the captured events (in order) and the current outbox rows, attached as the
+	 * reason on the ordering assertion so a CI failure shows the actual asynchronous, timing-dependent
+	 * sequence.
+	 */
+	@SuppressWarnings("unchecked")
+	private String describeState(List<TestOutboxEventListener.TestEvent> capturedEvents) {
+		String captured = capturedEvents.stream().map(TestOutboxEventListener.TestEvent::getMethod)
+		        .collect(Collectors.joining(", "));
+		Context.clearSession();
+		String outbox = ((List<OutboxEvent>) sessionFactory.getCurrentSession().createQuery("from OutboxEvent order by id")
+		        .list()).stream()
+		        .map(e -> e.getEventType() + "[status=" + e.getStatus() + ", errorCount=" + e.getErrorCount()
+		                + ", completedListeners=" + e.getCompletedListeners() + "]")
+		        .collect(Collectors.joining("; "));
+		return "\nCaptured (in order): [" + captured + "]\nOutbox rows: [" + outbox + "]\n";
+	}
+
 	private void waitForCapturedEvents(int count) throws InterruptedException {
 		long start = System.currentTimeMillis();
 		int capturedCount = 0;

api/src/test/java/org/openmrs/event/outbox/OutboxEventServiceConcurrencyIT.java

@@ -0,0 +1,94 @@
+/**
+ * This Source Code Form is subject to the terms of the Mozilla Public License,
+ * v. 2.0. If a copy of the MPL was not distributed with this file, You can
+ * obtain one at http://mozilla.org/MPL/2.0/. OpenMRS is also distributed under
+ * the terms of the Healthcare Disclaimer located at http://openmrs.org/license.
+ *
+ * Copyright (C) OpenMRS Inc. OpenMRS is a registered trademark and the OpenMRS
+ * graphic logo is a trademark of OpenMRS Inc.
+ */
+package org.openmrs.event.outbox;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.List;
+import java.util.UUID;
+
+import org.hibernate.SessionFactory;
+import org.junit.jupiter.api.Test;
+import org.openmrs.api.context.Context;
+import org.openmrs.test.jupiter.BaseContextSensitiveNonTransactionalTest;
+import org.springframework.beans.factory.annotation.Autowired;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+/**
+ * Guards the transactional outbox against a stale-read race: a poller reads a pending event
+ * (populating the Hibernate first-level cache) before claiming it, while another poller partially
+ * processes and releases the same event in between. The claiming poller must resume from the
+ * committed progress rather than its stale cached read, otherwise it re-runs listeners that were
+ * already recorded as completed.
+ */
+public class OutboxEventServiceConcurrencyIT extends BaseContextSensitiveNonTransactionalTest {
+
+	@Autowired
+	private OutboxEventService service;
+
+	@Autowired
+	private SessionFactory sessionFactory;
+
+	@Test
+	public void lockEventForProcessing_shouldObserveProgressCommittedByAnotherHandlerAfterThisHandlersRead()
+	        throws Exception {
+		// Given a PENDING event with no recorded progress
+		OutboxEvent seed = new OutboxEvent();
+		seed.setUuid(UUID.randomUUID().toString());
+		seed.setEventType("TEST_EVENT");
+		seed.setPayload("{\"test\":\"data\"}");
+		seed.setStatus(OutboxEvent.Status.PENDING);
+		service.saveOutboxEvent(seed);
+		final Integer id = seed.getId();
+		Context.clearSession();
+
+		// And this handler has already read it before claiming it (exactly as the poller does via
+		// getProcessingAndPendingEvents()), which puts the null-progress state into the first-level cache
+		OutboxEvent handle = sessionFactory.getCurrentSession().get(OutboxEvent.class, id);
+		assertNull(handle.getCompletedListeners());
+
+		// When another handler commits progress on the same row from a separate session
+		final String progress = "testOutboxEventListener.onSomething";
+		runInSeparateSession(() -> {
+			OutboxEvent inOtherSession = sessionFactory.getCurrentSession().get(OutboxEvent.class, id);
+			inOtherSession.setCompletedListeners(progress);
+			service.saveOutboxEvent(inOtherSession);
+		});
+
+		// Then claiming the event must reflect the committed progress, not this handler's stale cached read
+		boolean claimed = service.lockEventForProcessing(handle);
+		assertTrue(claimed, "the PENDING event should be claimable");
+		assertEquals(progress, handle.getCompletedListeners(),
+		    "lockEventForProcessing must refresh the committed progress, not return the stale cached read");
+	}
+
+	private void runInSeparateSession(Runnable action) throws InterruptedException {
+		List<Throwable> errors = Collections.synchronizedList(new ArrayList<>());
+		Thread thread = new Thread(() -> {
+			Context.openSession();
+			try {
+				Context.authenticate(getCredentials());
+				action.run();
+			} catch (Throwable t) {
+				errors.add(t);
+			} finally {
+				Context.closeSession();
+			}
+		});
+		thread.start();
+		thread.join();
+		if (!errors.isEmpty()) {
+			throw new RuntimeException("action in separate session failed", errors.get(0));
+		}
+	}
+}

api/src/test/java/org/openmrs/event/outbox/OutboxEventServiceTest.java

@@ -150,6 +150,8 @@ public void lockEventForProcessing_shouldClaimPendingEventAndReturnTrue() {
 
 		boolean locked = service.lockEventForProcessing(event);
 		assertTrue(locked, "Should successfully lock the PENDING event");
+		assertEquals(OutboxEvent.Status.PROCESSING, event.getStatus(),
+		    "The passed instance should reflect the claimed status");
 
 		OutboxEvent updated = sessionFactory.getCurrentSession().find(OutboxEvent.class, event.getId());
 		assertEquals(OutboxEvent.Status.PROCESSING, updated.getStatus());