fix(data): fire transaction synchronizations, and scope them to their transaction #9
+24
-3
@@ -53,9 +53,30 @@ public final class ResourceRegistry {
|
||||
SYNCHRONIZATIONS.get().add(Objects.requireNonNull(sync));
|
||||
}
|
||||
|
||||
public static void fireSynchronizations(TxOutcome outcome) {
|
||||
List<TxSynchronization> syncs = List.copyOf(SYNCHRONIZATIONS.get());
|
||||
SYNCHRONIZATIONS.get().clear();
|
||||
/**
|
||||
* How many synchronizations are registered right now — captured by a transaction manager when
|
||||
* it opens a new transaction, and handed back to {@link #fireSynchronizations} on completion
|
||||
* so that transaction only fires its own. See there for why that matters.
|
||||
*/
|
||||
public static int synchronizationCount() {
|
||||
return SYNCHRONIZATIONS.get().size();
|
||||
}
|
||||
|
||||
/**
|
||||
* Fires (and removes) the synchronizations registered from {@code fromIndex} onward — the ones
|
||||
* belonging to the transaction now completing. Everything before that index was registered by
|
||||
* an enclosing transaction that is merely <em>suspended</em>, not finished: a REQUIRES_NEW
|
||||
* inner transaction sets its own baseline, so committing it no longer drags the outer's
|
||||
* pending callbacks along — which fired them early, and with the inner transaction's outcome,
|
||||
* for an outer transaction that might still roll back.
|
||||
*/
|
||||
public static void fireSynchronizations(TxOutcome outcome, int fromIndex) {
|
||||
List<TxSynchronization> pending = SYNCHRONIZATIONS.get();
|
||||
if (fromIndex >= pending.size()) {
|
||||
return;
|
||||
}
|
||||
List<TxSynchronization> syncs = List.copyOf(pending.subList(fromIndex, pending.size()));
|
||||
pending.subList(fromIndex, pending.size()).clear();
|
||||
for (TxSynchronization sync : syncs) {
|
||||
if (outcome == TxOutcome.COMMITTED) {
|
||||
sync.afterCommit();
|
||||
|
||||
+22
-6
@@ -43,6 +43,9 @@ public class HibernateTxManager implements TxManager {
|
||||
}
|
||||
|
||||
private TxStatus beginNew(TxDefinition definition, HibernateTxStatus suspended) {
|
||||
// Anything already registered belongs to an enclosing transaction this one is nested
|
||||
// inside (or suspended over) — see ResourceRegistry#fireSynchronizations.
|
||||
int synchronizationBaseline = ResourceRegistry.synchronizationCount();
|
||||
Session s = sf.openSession();
|
||||
boolean bound = false;
|
||||
try {
|
||||
@@ -56,7 +59,8 @@ public class HibernateTxManager implements TxManager {
|
||||
true,
|
||||
definition.readOnly(),
|
||||
suspended,
|
||||
new HibernateTxStatus.RollbackMarker()
|
||||
new HibernateTxStatus.RollbackMarker(),
|
||||
synchronizationBaseline
|
||||
);
|
||||
ResourceRegistry.bind(HIBERNATE_STATUS_KEY, status);
|
||||
bound = true;
|
||||
@@ -80,12 +84,15 @@ public class HibernateTxManager implements TxManager {
|
||||
if (definition.readOnly() && !existing.isReadOnly()) {
|
||||
throw new TxException("Cannot join read-write tx as read-only");
|
||||
}
|
||||
// Baseline 0 is never read: a joined status isn't a new transaction, so commit()/rollback()
|
||||
// hand it straight back to the transaction it joined without firing anything.
|
||||
return new HibernateTxStatus(
|
||||
existing.session(),
|
||||
false,
|
||||
definition.readOnly(),
|
||||
null,
|
||||
existing.rollbackMarker()
|
||||
existing.rollbackMarker(),
|
||||
0
|
||||
);
|
||||
}
|
||||
|
||||
@@ -94,7 +101,7 @@ public class HibernateTxManager implements TxManager {
|
||||
}
|
||||
|
||||
private TxStatus noOp(TxDefinition definition, HibernateTxStatus suspended) {
|
||||
return new HibernateTxStatus(null, false, definition.readOnly(), suspended, new HibernateTxStatus.RollbackMarker());
|
||||
return new HibernateTxStatus(null, false, definition.readOnly(), suspended, new HibernateTxStatus.RollbackMarker(), 0);
|
||||
}
|
||||
|
||||
private HibernateTxStatus suspendIfNeeded() {
|
||||
@@ -114,7 +121,7 @@ public class HibernateTxManager implements TxManager {
|
||||
cleanupIfIdle();
|
||||
return;
|
||||
}
|
||||
TxOutcome outcome;
|
||||
TxOutcome outcome = null;
|
||||
try {
|
||||
if (s.isRollbackOnly() && s.session().getTransaction().isActive()) {
|
||||
s.session().getTransaction().rollback();
|
||||
@@ -124,10 +131,17 @@ public class HibernateTxManager implements TxManager {
|
||||
outcome = TxOutcome.COMMITTED;
|
||||
}
|
||||
} finally {
|
||||
// Order is load-bearing, and getting it wrong is silent: cleanupIfIdle() calls
|
||||
// ResourceRegistry.cleanup(), which removes the very ThreadLocal list of
|
||||
// synchronizations still waiting to be fired — firing afterwards saw a freshly
|
||||
// initialized empty list and dropped every callback on the floor. cleanupAndResume()
|
||||
// still has to come first, so a synchronization that opens its own transaction
|
||||
// (Registry#reload() in Pathway does) starts a fresh one instead of joining the
|
||||
// session that just committed. rollback() below already had this order right.
|
||||
cleanupAndResume(s);
|
||||
if (outcome != null) ResourceRegistry.fireSynchronizations(outcome, s.synchronizationBaseline());
|
||||
cleanupIfIdle();
|
||||
}
|
||||
ResourceRegistry.fireSynchronizations(outcome);
|
||||
}
|
||||
|
||||
@Override
|
||||
@@ -143,9 +157,11 @@ public class HibernateTxManager implements TxManager {
|
||||
if (s.session().getTransaction().isActive()) {
|
||||
s.session().getTransaction().rollback();
|
||||
}
|
||||
ResourceRegistry.fireSynchronizations(TxOutcome.ROLLED_BACK);
|
||||
} finally {
|
||||
// Same order as commit(): unbind the session first so a callback opening its own
|
||||
// transaction gets a fresh one, fire before cleanupIfIdle() can drop the list.
|
||||
cleanupAndResume(s);
|
||||
ResourceRegistry.fireSynchronizations(TxOutcome.ROLLED_BACK, s.synchronizationBaseline());
|
||||
cleanupIfIdle();
|
||||
}
|
||||
}
|
||||
|
||||
+7
-1
@@ -13,19 +13,22 @@ class HibernateTxStatus implements TxStatus {
|
||||
private final boolean readOnly;
|
||||
private final HibernateTxStatus suspended;
|
||||
private final RollbackMarker rollbackMarker;
|
||||
private final int synchronizationBaseline;
|
||||
|
||||
HibernateTxStatus(
|
||||
Session session,
|
||||
boolean newTransaction,
|
||||
boolean readOnly,
|
||||
HibernateTxStatus suspended,
|
||||
RollbackMarker rollbackMarker
|
||||
RollbackMarker rollbackMarker,
|
||||
int synchronizationBaseline
|
||||
) {
|
||||
this.session = session;
|
||||
this.newTransaction = newTransaction;
|
||||
this.readOnly = readOnly;
|
||||
this.suspended = suspended;
|
||||
this.rollbackMarker = rollbackMarker;
|
||||
this.synchronizationBaseline = synchronizationBaseline;
|
||||
}
|
||||
|
||||
@Override public boolean isNewTransaction() { return newTransaction; }
|
||||
@@ -44,4 +47,7 @@ class HibernateTxStatus implements TxStatus {
|
||||
Session session() { return session; }
|
||||
HibernateTxStatus suspended() { return suspended; }
|
||||
RollbackMarker rollbackMarker() { return rollbackMarker; }
|
||||
|
||||
/** Index into {@code ResourceRegistry}'s synchronization list where this transaction's own callbacks start. */
|
||||
int synchronizationBaseline() { return synchronizationBaseline; }
|
||||
}
|
||||
|
||||
+195
@@ -0,0 +1,195 @@
|
||||
package dev.relism.flash.ext.data.hibernate;
|
||||
|
||||
import dev.relism.flash.ext.data.core.*;
|
||||
import org.hibernate.Session;
|
||||
import org.hibernate.SessionFactory;
|
||||
import org.junit.jupiter.api.AfterAll;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.BeforeAll;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
/**
|
||||
* Transaction synchronization semantics — the {@code afterCommit}/{@code afterRollback} callbacks
|
||||
* {@code Data#afterCommit} exposes, and the contract callers build on: "my callback runs once, on
|
||||
* the outermost commit, after the session is gone".
|
||||
*
|
||||
* <p>None of this was covered before, and the gap was not academic: {@link #afterCommit_fires_on_commit}
|
||||
* failed against the original {@code commit()}, which fired synchronizations only after
|
||||
* {@code cleanupIfIdle()} had already dropped the ThreadLocal list holding them — so every callback
|
||||
* was silently discarded, on every commit, with no error and no log. Downstream that meant an admin
|
||||
* write landing in Postgres while the in-memory cache it was supposed to refresh never heard about
|
||||
* it until the process restarted.
|
||||
*/
|
||||
class HibernateTxManagerSynchronizationTest {
|
||||
|
||||
static SessionFactory sf;
|
||||
static HibernateTxManager manager;
|
||||
|
||||
@BeforeAll
|
||||
static void setup() {
|
||||
sf = TestHelper.buildSessionFactory();
|
||||
manager = new HibernateTxManager(sf);
|
||||
}
|
||||
|
||||
@AfterAll
|
||||
static void teardown() {
|
||||
if (sf != null) {
|
||||
sf.close();
|
||||
}
|
||||
}
|
||||
|
||||
@AfterEach
|
||||
void cleanup() {
|
||||
ResourceRegistry.clear();
|
||||
}
|
||||
|
||||
/** Records which callbacks ran, in order — order is part of the contract, not just the fact they ran. */
|
||||
private static final class Recorder implements TxSynchronization {
|
||||
final List<String> calls = new ArrayList<>();
|
||||
|
||||
@Override public void afterCommit() { calls.add("afterCommit"); }
|
||||
@Override public void afterRollback() { calls.add("afterRollback"); }
|
||||
@Override public void afterCompletion(TxOutcome outcome) { calls.add("afterCompletion:" + outcome); }
|
||||
}
|
||||
|
||||
@Test
|
||||
void afterCommit_fires_on_commit() {
|
||||
Recorder recorder = new Recorder();
|
||||
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||
ResourceRegistry.addSynchronization(recorder);
|
||||
|
||||
manager.commit(tx);
|
||||
|
||||
assertEquals(List.of("afterCommit", "afterCompletion:COMMITTED"), recorder.calls);
|
||||
}
|
||||
|
||||
@Test
|
||||
void afterRollback_fires_on_rollback() {
|
||||
Recorder recorder = new Recorder();
|
||||
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||
ResourceRegistry.addSynchronization(recorder);
|
||||
|
||||
manager.rollback(tx);
|
||||
|
||||
assertEquals(List.of("afterRollback", "afterCompletion:ROLLED_BACK"), recorder.calls);
|
||||
}
|
||||
|
||||
/** A commit() call on a tx already marked rollback-only really rolls back — the callbacks must say so. */
|
||||
@Test
|
||||
void commit_of_a_rollback_only_tx_fires_the_rollback_callbacks() {
|
||||
Recorder recorder = new Recorder();
|
||||
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||
tx.markRollbackOnly();
|
||||
ResourceRegistry.addSynchronization(recorder);
|
||||
|
||||
manager.commit(tx);
|
||||
|
||||
assertEquals(List.of("afterRollback", "afterCompletion:ROLLED_BACK"), recorder.calls);
|
||||
}
|
||||
|
||||
/**
|
||||
* The load-bearing ordering detail: callbacks run <em>after</em> the committed session is
|
||||
* unbound, so a callback that opens its own transaction (a cache reload, an outbox drain)
|
||||
* gets a fresh one instead of silently joining the transaction that just committed.
|
||||
*/
|
||||
@Test
|
||||
void a_synchronization_may_open_its_own_transaction() {
|
||||
List<Session> sessionsSeen = new ArrayList<>();
|
||||
List<Boolean> wasNewTransaction = new ArrayList<>();
|
||||
|
||||
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||
Session committedSession = outer.resource(Session.class);
|
||||
ResourceRegistry.addSynchronization(new TxSynchronization() {
|
||||
@Override
|
||||
public void afterCommit() {
|
||||
TxStatus own = manager.begin(TxDefinition.DEFAULTS);
|
||||
sessionsSeen.add(own.resource(Session.class));
|
||||
wasNewTransaction.add(own.isNewTransaction());
|
||||
manager.commit(own);
|
||||
}
|
||||
});
|
||||
|
||||
manager.commit(outer);
|
||||
|
||||
assertEquals(1, sessionsSeen.size(), "the callback must have run");
|
||||
assertEquals(List.of(true), wasNewTransaction, "must start its own transaction, not join the committed one");
|
||||
assertNotSame(committedSession, sessionsSeen.get(0));
|
||||
}
|
||||
|
||||
/** A joined (REQUIRED) inner commit is not a real commit — callbacks wait for the outermost one. */
|
||||
@Test
|
||||
void a_joined_commit_defers_synchronizations_to_the_outermost_commit() {
|
||||
Recorder recorder = new Recorder();
|
||||
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||
TxStatus inner = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRED));
|
||||
ResourceRegistry.addSynchronization(recorder);
|
||||
|
||||
manager.commit(inner);
|
||||
assertEquals(List.of(), recorder.calls, "the joined commit did not commit anything yet");
|
||||
|
||||
manager.commit(outer);
|
||||
assertEquals(List.of("afterCommit", "afterCompletion:COMMITTED"), recorder.calls);
|
||||
}
|
||||
|
||||
/** Each callback belongs to one transaction: a second transaction must not re-run the first's. */
|
||||
@Test
|
||||
void synchronizations_do_not_leak_into_the_next_transaction() {
|
||||
Recorder recorder = new Recorder();
|
||||
TxStatus first = manager.begin(TxDefinition.DEFAULTS);
|
||||
ResourceRegistry.addSynchronization(recorder);
|
||||
manager.commit(first);
|
||||
recorder.calls.clear();
|
||||
|
||||
TxStatus second = manager.begin(TxDefinition.DEFAULTS);
|
||||
manager.commit(second);
|
||||
|
||||
assertEquals(List.of(), recorder.calls);
|
||||
}
|
||||
|
||||
/**
|
||||
* A REQUIRES_NEW inner transaction suspends the outer one; committing the inner must not drag
|
||||
* the still-pending outer transaction's callbacks along with it. They belong to a transaction
|
||||
* that has not committed — and may yet roll back, in which case firing {@code afterCommit} for
|
||||
* it would be a straight lie.
|
||||
*/
|
||||
@Test
|
||||
void a_requires_new_commit_leaves_the_suspended_transactions_synchronizations_alone() {
|
||||
Recorder outerSync = new Recorder();
|
||||
Recorder innerSync = new Recorder();
|
||||
|
||||
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||
ResourceRegistry.addSynchronization(outerSync);
|
||||
|
||||
TxStatus inner = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRES_NEW));
|
||||
ResourceRegistry.addSynchronization(innerSync);
|
||||
manager.commit(inner);
|
||||
|
||||
assertEquals(List.of("afterCommit", "afterCompletion:COMMITTED"), innerSync.calls);
|
||||
assertEquals(List.of(), outerSync.calls, "the outer transaction has not committed yet");
|
||||
|
||||
manager.commit(outer);
|
||||
assertEquals(List.of("afterCommit", "afterCompletion:COMMITTED"), outerSync.calls);
|
||||
}
|
||||
|
||||
/** A rolled-back inner REQUIRES_NEW must not fire the outer's callbacks either — same reason, opposite outcome. */
|
||||
@Test
|
||||
void a_requires_new_rollback_leaves_the_suspended_transactions_synchronizations_alone() {
|
||||
Recorder outerSync = new Recorder();
|
||||
|
||||
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||
ResourceRegistry.addSynchronization(outerSync);
|
||||
|
||||
TxStatus inner = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRES_NEW));
|
||||
manager.rollback(inner);
|
||||
|
||||
assertEquals(List.of(), outerSync.calls, "the outer transaction is still open");
|
||||
|
||||
manager.commit(outer);
|
||||
assertEquals(List.of("afterCommit", "afterCompletion:COMMITTED"), outerSync.calls);
|
||||
}
|
||||
}
|
||||
+40
@@ -65,4 +65,44 @@ class HibernateTxManagerTest {
|
||||
assertTrue(outer.isRollbackOnly());
|
||||
manager.rollback(outer);
|
||||
}
|
||||
|
||||
/** SUPPORTS without an active transaction yields a sessionless status: not a transaction, no session to hand out. */
|
||||
@Test
|
||||
void supports_without_active_transaction_is_a_sessionless_no_op() {
|
||||
TxStatus s = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.SUPPORTS));
|
||||
|
||||
assertFalse(s.isNewTransaction());
|
||||
assertThrows(IllegalStateException.class, () -> s.resource(Session.class));
|
||||
assertDoesNotThrow(() -> manager.commit(s));
|
||||
}
|
||||
|
||||
@Test
|
||||
void not_supported_suspends_the_active_transaction_and_restores_it_on_commit() {
|
||||
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||
Session outerSession = outer.resource(Session.class);
|
||||
|
||||
TxStatus suspended = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.NOT_SUPPORTED));
|
||||
assertFalse(suspended.isNewTransaction());
|
||||
assertThrows(IllegalStateException.class, () -> suspended.resource(Session.class));
|
||||
manager.commit(suspended);
|
||||
|
||||
TxStatus rejoined = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRED));
|
||||
assertSame(outerSession, rejoined.resource(Session.class), "the suspended transaction must be back");
|
||||
manager.rollback(outer);
|
||||
}
|
||||
|
||||
@Test
|
||||
void mandatory_without_active_transaction_is_rejected() {
|
||||
assertThrows(IllegalStateException.class,
|
||||
() -> manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.MANDATORY)));
|
||||
}
|
||||
|
||||
/** A read-only join onto a read-write transaction is a contract violation, not a silent downgrade. */
|
||||
@Test
|
||||
void read_only_cannot_join_a_read_write_transaction() {
|
||||
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||
assertThrows(TxException.class,
|
||||
() -> manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRED).asReadOnly()));
|
||||
manager.rollback(outer);
|
||||
}
|
||||
}
|
||||
|
||||
+21
-8
@@ -41,6 +41,9 @@ public class JdbcTxManager implements TxManager {
|
||||
|
||||
private TxStatus beginNew(TxDefinition definition) {
|
||||
Connection conn = null;
|
||||
// Anything already registered belongs to an enclosing transaction this one is nested
|
||||
// inside (or suspended over) — see ResourceRegistry#fireSynchronizations.
|
||||
int synchronizationBaseline = ResourceRegistry.synchronizationCount();
|
||||
JdbcTxStatus suspended = suspendIfNeeded();
|
||||
boolean bound = false;
|
||||
try {
|
||||
@@ -55,7 +58,8 @@ public class JdbcTxManager implements TxManager {
|
||||
true,
|
||||
definition.readOnly(),
|
||||
suspended,
|
||||
new JdbcTxStatus.RollbackMarker()
|
||||
new JdbcTxStatus.RollbackMarker(),
|
||||
synchronizationBaseline
|
||||
);
|
||||
ResourceRegistry.bind(JDBC_STATUS_KEY, status);
|
||||
bound = true;
|
||||
@@ -76,12 +80,15 @@ public class JdbcTxManager implements TxManager {
|
||||
if (definition.readOnly() && !existing.isReadOnly()) {
|
||||
throw new TxException("Cannot join read-write tx as read-only");
|
||||
}
|
||||
// Baseline 0 is never read: a joined status isn't a new transaction, so commit()/rollback()
|
||||
// hand it straight back to the transaction it joined without firing anything.
|
||||
return new JdbcTxStatus(
|
||||
existing.connection(),
|
||||
false,
|
||||
definition.readOnly(),
|
||||
null,
|
||||
existing.rollbackMarker()
|
||||
existing.rollbackMarker(),
|
||||
0
|
||||
);
|
||||
}
|
||||
|
||||
@@ -90,7 +97,7 @@ public class JdbcTxManager implements TxManager {
|
||||
}
|
||||
|
||||
private TxStatus noOp(TxDefinition definition, JdbcTxStatus suspended) {
|
||||
return new JdbcTxStatus(null, false, definition.readOnly(), suspended, new JdbcTxStatus.RollbackMarker());
|
||||
return new JdbcTxStatus(null, false, definition.readOnly(), suspended, new JdbcTxStatus.RollbackMarker(), 0);
|
||||
}
|
||||
|
||||
private JdbcTxStatus suspendIfNeeded() {
|
||||
@@ -110,18 +117,23 @@ public class JdbcTxManager implements TxManager {
|
||||
cleanupIfIdle();
|
||||
return;
|
||||
}
|
||||
TxOutcome outcome = null;
|
||||
try {
|
||||
if (s.isRollbackOnly()) {
|
||||
s.connection().rollback();
|
||||
ResourceRegistry.fireSynchronizations(TxOutcome.ROLLED_BACK);
|
||||
return;
|
||||
}
|
||||
outcome = TxOutcome.ROLLED_BACK;
|
||||
} else {
|
||||
s.connection().commit();
|
||||
ResourceRegistry.fireSynchronizations(TxOutcome.COMMITTED);
|
||||
outcome = TxOutcome.COMMITTED;
|
||||
}
|
||||
} catch (SQLException e) {
|
||||
throw new TxException(e);
|
||||
} finally {
|
||||
// Unbind the connection before firing, so a callback that opens its own transaction
|
||||
// gets a fresh one instead of joining the connection that just committed — and fire
|
||||
// before cleanupIfIdle(), whose ResourceRegistry.cleanup() drops the pending list.
|
||||
cleanupAndResume(s);
|
||||
if (outcome != null) ResourceRegistry.fireSynchronizations(outcome, s.synchronizationBaseline());
|
||||
cleanupIfIdle();
|
||||
}
|
||||
}
|
||||
@@ -137,11 +149,12 @@ public class JdbcTxManager implements TxManager {
|
||||
}
|
||||
try {
|
||||
s.connection().rollback();
|
||||
ResourceRegistry.fireSynchronizations(TxOutcome.ROLLED_BACK);
|
||||
} catch (SQLException e) {
|
||||
throw new TxException(e);
|
||||
} finally {
|
||||
// Same order as commit() above, for the same two reasons.
|
||||
cleanupAndResume(s);
|
||||
ResourceRegistry.fireSynchronizations(TxOutcome.ROLLED_BACK, s.synchronizationBaseline());
|
||||
cleanupIfIdle();
|
||||
}
|
||||
}
|
||||
|
||||
+13
-3
@@ -3,7 +3,6 @@ package dev.relism.flash.ext.data.jdbc;
|
||||
import dev.relism.flash.ext.data.core.TxStatus;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.util.Objects;
|
||||
|
||||
class JdbcTxStatus implements TxStatus {
|
||||
static final class RollbackMarker {
|
||||
@@ -15,19 +14,27 @@ class JdbcTxStatus implements TxStatus {
|
||||
private final boolean readOnly;
|
||||
private final JdbcTxStatus suspended;
|
||||
private final RollbackMarker rollbackMarker;
|
||||
private final int synchronizationBaseline;
|
||||
|
||||
// No requireNonNull on the connection: a SUPPORTS-without-a-transaction or a NOT_SUPPORTED
|
||||
// status is deliberately connectionless (see JdbcTxManager#noOp), and rejecting null here
|
||||
// turned both of those propagations into an NPE at begin() — the Hibernate manager has
|
||||
// always allowed it. resource() reports the real mistake, asking a connectionless status for
|
||||
// its connection, where it can name it.
|
||||
JdbcTxStatus(
|
||||
Connection connection,
|
||||
boolean newTransaction,
|
||||
boolean readOnly,
|
||||
JdbcTxStatus suspended,
|
||||
RollbackMarker rollbackMarker
|
||||
RollbackMarker rollbackMarker,
|
||||
int synchronizationBaseline
|
||||
) {
|
||||
this.connection = Objects.requireNonNull(connection);
|
||||
this.connection = connection;
|
||||
this.newTransaction = newTransaction;
|
||||
this.readOnly = readOnly;
|
||||
this.suspended = suspended;
|
||||
this.rollbackMarker = rollbackMarker;
|
||||
this.synchronizationBaseline = synchronizationBaseline;
|
||||
}
|
||||
|
||||
@Override public boolean isNewTransaction() { return newTransaction; }
|
||||
@@ -46,4 +53,7 @@ class JdbcTxStatus implements TxStatus {
|
||||
Connection connection() { return connection; }
|
||||
JdbcTxStatus suspended() { return suspended; }
|
||||
RollbackMarker rollbackMarker() { return rollbackMarker; }
|
||||
|
||||
/** Index into {@code ResourceRegistry}'s synchronization list where this transaction's own callbacks start. */
|
||||
int synchronizationBaseline() { return synchronizationBaseline; }
|
||||
}
|
||||
|
||||
+158
@@ -0,0 +1,158 @@
|
||||
package dev.relism.flash.ext.data.jdbc;
|
||||
|
||||
import dev.relism.flash.ext.data.core.*;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import java.sql.Connection;
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
/**
|
||||
* Transaction synchronization semantics for the JDBC manager — the same contract {@code
|
||||
* HibernateTxManagerSynchronizationTest} pins down for the Hibernate one, kept deliberately
|
||||
* parallel: the two managers are interchangeable behind {@code TxManager}, so a callback must not
|
||||
* observe a different lifecycle depending on which one is installed.
|
||||
*/
|
||||
class JdbcTxManagerSynchronizationTest {
|
||||
|
||||
private final JdbcTxManager manager = new JdbcTxManager(new TestDataSource());
|
||||
|
||||
@AfterEach
|
||||
void cleanup() {
|
||||
ResourceRegistry.clear();
|
||||
}
|
||||
|
||||
/** Records which callbacks ran, in order — order is part of the contract, not just the fact they ran. */
|
||||
private static final class Recorder implements TxSynchronization {
|
||||
final List<String> calls = new ArrayList<>();
|
||||
|
||||
@Override public void afterCommit() { calls.add("afterCommit"); }
|
||||
@Override public void afterRollback() { calls.add("afterRollback"); }
|
||||
@Override public void afterCompletion(TxOutcome outcome) { calls.add("afterCompletion:" + outcome); }
|
||||
}
|
||||
|
||||
@Test
|
||||
void afterCommit_fires_on_commit() {
|
||||
Recorder recorder = new Recorder();
|
||||
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||
ResourceRegistry.addSynchronization(recorder);
|
||||
|
||||
manager.commit(tx);
|
||||
|
||||
assertEquals(List.of("afterCommit", "afterCompletion:COMMITTED"), recorder.calls);
|
||||
}
|
||||
|
||||
@Test
|
||||
void afterRollback_fires_on_rollback() {
|
||||
Recorder recorder = new Recorder();
|
||||
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||
ResourceRegistry.addSynchronization(recorder);
|
||||
|
||||
manager.rollback(tx);
|
||||
|
||||
assertEquals(List.of("afterRollback", "afterCompletion:ROLLED_BACK"), recorder.calls);
|
||||
}
|
||||
|
||||
@Test
|
||||
void commit_of_a_rollback_only_tx_fires_the_rollback_callbacks() {
|
||||
Recorder recorder = new Recorder();
|
||||
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
|
||||
tx.markRollbackOnly();
|
||||
ResourceRegistry.addSynchronization(recorder);
|
||||
|
||||
manager.commit(tx);
|
||||
|
||||
assertEquals(List.of("afterRollback", "afterCompletion:ROLLED_BACK"), recorder.calls);
|
||||
}
|
||||
|
||||
/** Callbacks run after the committed connection is unbound, so opening a transaction gets a fresh one. */
|
||||
@Test
|
||||
void a_synchronization_may_open_its_own_transaction() {
|
||||
List<Connection> connectionsSeen = new ArrayList<>();
|
||||
List<Boolean> wasNewTransaction = new ArrayList<>();
|
||||
|
||||
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||
Connection committedConnection = outer.resource(Connection.class);
|
||||
ResourceRegistry.addSynchronization(new TxSynchronization() {
|
||||
@Override
|
||||
public void afterCommit() {
|
||||
TxStatus own = manager.begin(TxDefinition.DEFAULTS);
|
||||
connectionsSeen.add(own.resource(Connection.class));
|
||||
wasNewTransaction.add(own.isNewTransaction());
|
||||
manager.commit(own);
|
||||
}
|
||||
});
|
||||
|
||||
manager.commit(outer);
|
||||
|
||||
assertEquals(1, connectionsSeen.size(), "the callback must have run");
|
||||
assertEquals(List.of(true), wasNewTransaction, "must start its own transaction, not join the committed one");
|
||||
assertNotSame(committedConnection, connectionsSeen.get(0));
|
||||
}
|
||||
|
||||
@Test
|
||||
void a_joined_commit_defers_synchronizations_to_the_outermost_commit() {
|
||||
Recorder recorder = new Recorder();
|
||||
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||
TxStatus inner = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRED));
|
||||
ResourceRegistry.addSynchronization(recorder);
|
||||
|
||||
manager.commit(inner);
|
||||
assertEquals(List.of(), recorder.calls, "the joined commit did not commit anything yet");
|
||||
|
||||
manager.commit(outer);
|
||||
assertEquals(List.of("afterCommit", "afterCompletion:COMMITTED"), recorder.calls);
|
||||
}
|
||||
|
||||
@Test
|
||||
void synchronizations_do_not_leak_into_the_next_transaction() {
|
||||
Recorder recorder = new Recorder();
|
||||
TxStatus first = manager.begin(TxDefinition.DEFAULTS);
|
||||
ResourceRegistry.addSynchronization(recorder);
|
||||
manager.commit(first);
|
||||
recorder.calls.clear();
|
||||
|
||||
TxStatus second = manager.begin(TxDefinition.DEFAULTS);
|
||||
manager.commit(second);
|
||||
|
||||
assertEquals(List.of(), recorder.calls);
|
||||
}
|
||||
|
||||
@Test
|
||||
void a_requires_new_commit_leaves_the_suspended_transactions_synchronizations_alone() {
|
||||
Recorder outerSync = new Recorder();
|
||||
Recorder innerSync = new Recorder();
|
||||
|
||||
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||
ResourceRegistry.addSynchronization(outerSync);
|
||||
|
||||
TxStatus inner = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRES_NEW));
|
||||
ResourceRegistry.addSynchronization(innerSync);
|
||||
manager.commit(inner);
|
||||
|
||||
assertEquals(List.of("afterCommit", "afterCompletion:COMMITTED"), innerSync.calls);
|
||||
assertEquals(List.of(), outerSync.calls, "the outer transaction has not committed yet");
|
||||
|
||||
manager.commit(outer);
|
||||
assertEquals(List.of("afterCommit", "afterCompletion:COMMITTED"), outerSync.calls);
|
||||
}
|
||||
|
||||
@Test
|
||||
void a_requires_new_rollback_leaves_the_suspended_transactions_synchronizations_alone() {
|
||||
Recorder outerSync = new Recorder();
|
||||
|
||||
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||
ResourceRegistry.addSynchronization(outerSync);
|
||||
|
||||
TxStatus inner = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRES_NEW));
|
||||
manager.rollback(inner);
|
||||
|
||||
assertEquals(List.of(), outerSync.calls, "the outer transaction is still open");
|
||||
|
||||
manager.commit(outer);
|
||||
assertEquals(List.of("afterCommit", "afterCompletion:COMMITTED"), outerSync.calls);
|
||||
}
|
||||
}
|
||||
+30
-47
@@ -4,15 +4,12 @@ import dev.relism.flash.ext.data.core.*;
|
||||
import org.junit.jupiter.api.AfterEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
import java.sql.Connection;
|
||||
import java.sql.DriverManager;
|
||||
import java.sql.SQLException;
|
||||
|
||||
import static org.junit.jupiter.api.Assertions.*;
|
||||
|
||||
class JdbcTxManagerTest {
|
||||
private final JdbcTxManager manager = new JdbcTxManager(dataSource());
|
||||
private final JdbcTxManager manager = new JdbcTxManager(new TestDataSource());
|
||||
|
||||
@AfterEach
|
||||
void cleanup() {
|
||||
@@ -53,52 +50,38 @@ class JdbcTxManagerTest {
|
||||
manager.rollback(outer);
|
||||
}
|
||||
|
||||
private static DataSource dataSource() {
|
||||
return new DataSource() {
|
||||
@Override
|
||||
public Connection getConnection() throws SQLException {
|
||||
return DriverManager.getConnection("jdbc:h2:mem:tx-jdbc;DB_CLOSE_DELAY=-1");
|
||||
/**
|
||||
* SUPPORTS without an active transaction yields a connectionless status — it must not be a
|
||||
* transaction, and asking it for a connection must say so rather than NPE. Both propagations
|
||||
* that produce one used to throw {@link NullPointerException} straight out of {@code begin()}.
|
||||
*/
|
||||
@Test
|
||||
void supports_without_active_transaction_is_a_connectionless_no_op() {
|
||||
TxStatus s = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.SUPPORTS));
|
||||
|
||||
assertFalse(s.isNewTransaction());
|
||||
assertThrows(IllegalStateException.class, () -> s.resource(Connection.class));
|
||||
assertDoesNotThrow(() -> manager.commit(s));
|
||||
}
|
||||
|
||||
@Override
|
||||
public Connection getConnection(String username, String password) throws SQLException {
|
||||
return DriverManager.getConnection("jdbc:h2:mem:tx-jdbc;DB_CLOSE_DELAY=-1", username, password);
|
||||
@Test
|
||||
void not_supported_suspends_the_active_transaction_and_restores_it_on_commit() {
|
||||
TxStatus outer = manager.begin(TxDefinition.DEFAULTS);
|
||||
Connection outerConnection = outer.resource(Connection.class);
|
||||
|
||||
TxStatus suspended = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.NOT_SUPPORTED));
|
||||
assertFalse(suspended.isNewTransaction());
|
||||
assertThrows(IllegalStateException.class, () -> suspended.resource(Connection.class));
|
||||
manager.commit(suspended);
|
||||
|
||||
TxStatus rejoined = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.REQUIRED));
|
||||
assertSame(outerConnection, rejoined.resource(Connection.class), "the suspended transaction must be back");
|
||||
manager.rollback(outer);
|
||||
}
|
||||
|
||||
@Override
|
||||
public <T> T unwrap(Class<T> iface) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isWrapperFor(Class<?> iface) {
|
||||
return false;
|
||||
}
|
||||
|
||||
@Override
|
||||
public java.io.PrintWriter getLogWriter() {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setLogWriter(java.io.PrintWriter out) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public void setLoginTimeout(int seconds) {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
|
||||
@Override
|
||||
public int getLoginTimeout() {
|
||||
return 0;
|
||||
}
|
||||
|
||||
@Override
|
||||
public java.util.logging.Logger getParentLogger() {
|
||||
throw new UnsupportedOperationException();
|
||||
}
|
||||
};
|
||||
@Test
|
||||
void mandatory_without_active_transaction_is_rejected() {
|
||||
assertThrows(IllegalStateException.class,
|
||||
() -> manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.MANDATORY)));
|
||||
}
|
||||
}
|
||||
|
||||
+37
@@ -0,0 +1,37 @@
|
||||
package dev.relism.flash.ext.data.jdbc;
|
||||
|
||||
import javax.sql.DataSource;
|
||||
import java.io.PrintWriter;
|
||||
import java.sql.Connection;
|
||||
import java.sql.DriverManager;
|
||||
import java.sql.SQLException;
|
||||
import java.util.logging.Logger;
|
||||
|
||||
/**
|
||||
* A bare in-memory H2 {@link DataSource} — one fresh connection per {@code getConnection()}, which
|
||||
* is all {@code JdbcTxManager} needs to exercise real commit/rollback and suspension. Every method
|
||||
* outside the two {@code getConnection} overloads throws: nothing under test calls them, and a
|
||||
* loud failure beats a silent stub if that ever changes.
|
||||
*/
|
||||
final class TestDataSource implements DataSource {
|
||||
|
||||
static final String URL = "jdbc:h2:mem:tx-jdbc;DB_CLOSE_DELAY=-1";
|
||||
|
||||
@Override
|
||||
public Connection getConnection() throws SQLException {
|
||||
return DriverManager.getConnection(URL);
|
||||
}
|
||||
|
||||
@Override
|
||||
public Connection getConnection(String username, String password) throws SQLException {
|
||||
return DriverManager.getConnection(URL, username, password);
|
||||
}
|
||||
|
||||
@Override public <T> T unwrap(Class<T> iface) { throw new UnsupportedOperationException(); }
|
||||
@Override public boolean isWrapperFor(Class<?> iface) { return false; }
|
||||
@Override public PrintWriter getLogWriter() { throw new UnsupportedOperationException(); }
|
||||
@Override public void setLogWriter(PrintWriter out) { throw new UnsupportedOperationException(); }
|
||||
@Override public void setLoginTimeout(int seconds) { throw new UnsupportedOperationException(); }
|
||||
@Override public int getLoginTimeout() { return 0; }
|
||||
@Override public Logger getParentLogger() { throw new UnsupportedOperationException(); }
|
||||
}
|
||||
Reference in New Issue
Block a user