Compare commits

4 Commits
Author SHA1 Message Date
Relism 8f1f30b973 Merge pull request 'fix(data): fire transaction synchronizations, and scope them to their transaction' (#9) from fix/tx-synchronizations into master
Publish Maven packages / publish (push) Successful in 1m53s
Reviewed-on: #9
2026-08-12 23:37:36 +00:00
Zakaria El OrcheandClaude Opus 5 c5179146c0 docs(data): write the data-layer docs in English
The three data modules' docs were in Italian, so the synchronization contract
added in the previous commit went in as Italian too, to match its file. English
is the project's language for docs, comments and READMEs alike, and a file half
in each is worse than either — so all three are translated, not just the new
section.

Content is otherwise unchanged, except the "synchronizations run on
commit/rollback" line in the two backend READMEs, which was vague before and is
now accurate about which hook sees the session/connection still bound, pointing
at flash-ext-data-core's README for the full contract.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-12 23:36:11 +00:00
Zakaria El OrcheandClaude Opus 5 0194470c1f feat(data): actually invoke TxSynchronization.beforeCommit, and document the contract
beforeCommit(boolean) has been part of the TxSynchronization API from the
start and was invoked by nothing, in either manager — anyone implementing it
got silence, the same failure class as the dropped afterCommit callbacks in
the previous commit. Left out of that one because fixing it is a design
decision rather than a restored behaviour; this is that decision, written
down.

It now runs immediately before the real commit, with the transaction still
active and its session/connection still bound, which is the whole reason to
have a hook on this side of the commit: it can still write through the same
resource and land in the same atomic unit. Skipped when the transaction is
already rollback-only, since there is no commit to precede.

Throwing from it vetoes the commit: the transaction rolls back, the surviving
callbacks hear ROLLED_BACK, and the exception propagates. Without that, a hook
running before the commit would be strictly less useful than one running
after. A commit that fails on its own now takes the same path instead of
completing silently with no callback at all, and a rollback that also fails is
attached as a suppressed exception rather than replacing the one that explains
the failure.

Documented on the interface itself and in flash-ext-data-core/docs/README.md:
which hook sits on which side of the commit, what each may still touch, what
throwing does, and the per-transaction scoping rule from the previous commit.

Tests: 6 more (3 per manager) — runs inside the transaction with the resource
still bound, receives the read-only flag, skipped on a rollback-only
transaction, and vetoes the commit when it throws. 37 across the two managers
now, all green, reactor verify passes the 80% Jacoco gate.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-12 23:33:05 +00:00
Zakaria El OrcheandClaude Opus 5 7299490d0d fix(data): fire transaction synchronizations, and scope them to their transaction
Two defects in the same mechanism, both silent.

1. HibernateTxManager#commit fired synchronizations *after* its finally block,
   where cleanupIfIdle() had already called ResourceRegistry.cleanup() and
   removed the ThreadLocal list holding them. fireSynchronizations() then read
   a freshly initialized empty list and did nothing. No afterCommit callback
   had ever run on the Hibernate path: no exception, no log, just silence.
   rollback() twenty lines below had the order right, which is what makes this
   an ordering slip rather than a design choice.

   Found in production, from the far end: an admin write landed in Postgres
   while the in-memory cache it was registered to refresh never heard about it,
   so the change only took effect when the process restarted and re-read the
   database at boot.

2. Synchronizations were a single flat per-thread list, fired from index 0 by
   whichever transaction completed first. A REQUIRES_NEW inner transaction
   therefore fired the *suspended* outer transaction's callbacks too — early,
   with the inner transaction's outcome, for a transaction that might still
   roll back. Each new transaction now records how many synchronizations were
   already registered when it began, and fires only its own tail.

Both managers get the fix and the same callback ordering: unbind the session or
connection first, so a callback that opens its own transaction (a cache reload,
an outbox drain) gets a fresh one instead of joining the transaction that just
committed, then fire, then clean up.

Also fixes JdbcTxStatus rejecting a null connection, which turned the two
propagations that deliberately produce a connectionless status — SUPPORTS with
no active transaction, and NOT_SUPPORTED — into an NPE inside begin(). The
Hibernate manager always allowed it, and resource() already reports the real
mistake with a message that names it.

Tests: 16 new across the two managers, kept deliberately parallel since the two
are interchangeable behind TxManager — synchronization firing, ordering,
per-transaction scoping, callbacks opening their own transaction, and the
previously untested SUPPORTS/NOT_SUPPORTED/MANDATORY propagations. Nothing
covered afterCommit before, which is how both defects shipped.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
2026-08-12 23:23:47 +00:00
14 changed files with 946 additions and 180 deletions
@@ -1,53 +1,93 @@
# flash-ext-data-core
Core comune per il layer dati di Flash.
Shared core for Flash's data layer.
## Scopo
## Purpose
Questo modulo definisce il contratto transazionale condiviso tra le implementazioni backend.
Non parla con Hibernate o JDBC direttamente: espone solo astrazioni e un runtime minimale.
This module defines the transactional contract shared across backend implementations. It does not
talk to Hibernate or JDBC directly: it exposes abstractions and a minimal runtime, nothing else.
## Componenti
## Components
- `TxDefinition`: metadata immutabile della transazione.
- `TxStatus`: stato runtime restituito dal manager.
- `TxManager`: contratto per `begin`, `commit`, `rollback`.
- `Tx`: orchestration runtime e stack transazionale per thread.
- `ResourceRegistry`: storage thread-local di risorse e synchronizations.
- `Repository<T, ID>`: base repository auto-transazionale.
- `Spec<T>`: predicato componibile.
- `Query<T>`: oggetto query con spec, sort e paging.
- `SpecBuilder<T>`: DSL fluente per costruire spec tipizzate.
- `RepositorySupport<T, ID>`: helper interno condiviso.
- `TransactionPropagation`: semantica di propagazione.
- `TransactionIsolation`: livello di isolamento.
- `TxSynchronization`: hook lifecycle.
- `TxDefinition`: immutable transaction metadata.
- `TxStatus`: runtime state returned by the manager.
- `TxManager`: the `begin`/`commit`/`rollback` contract.
- `Tx`: runtime orchestration and the per-thread transaction stack.
- `ResourceRegistry`: thread-local storage for resources and synchronizations.
- `Repository<T, ID>`: self-transactional base repository.
- `Spec<T>`: composable predicate.
- `Query<T>`: query object carrying spec, sort and paging.
- `SpecBuilder<T>`: fluent DSL for building typed specs.
- `RepositorySupport<T, ID>`: shared internal helper.
- `TransactionPropagation`: propagation semantics.
- `TransactionIsolation`: isolation level.
- `TxSynchronization`: lifecycle hooks (see below).
## Modello di esecuzione
## Execution model
Il flusso è:
The flow is:
1. `Tx.call(definition, work)` chiama `TxManager.begin(definition)`.
2. Il `TxManager` crea un `TxStatus` backend-specific.
3. Lo status viene pushato nello stack thread-local.
4. Il lavoro usa `Tx.resource(Class)` per ottenere la risorsa corrente.
5. A fine lavoro `Tx` decide tra `commit` e `rollback`.
6. Lo stack viene poppato e il thread-local viene pulito se vuoto.
1. `Tx.call(definition, work)` calls `TxManager.begin(definition)`.
2. The `TxManager` creates a backend-specific `TxStatus`.
3. The status is pushed onto the thread-local stack.
4. The work uses `Tx.resource(Class)` to obtain the current resource.
5. When the work ends, `Tx` chooses between `commit` and `rollback`.
6. The stack is popped, and the thread-local is cleared once it is empty.
## Propagation supportata
## Supported propagation
- `REQUIRED`: usa la tx attiva oppure ne apre una nuova.
- `REQUIRES_NEW`: sospende la tx corrente e apre una nuova tx.
- `SUPPORTS`: se esiste una tx attiva si aggancia, altrimenti esegue senza tx.
- `NOT_SUPPORTED`: sospende la tx corrente ed esegue senza tx.
- `MANDATORY`: richiede una tx attiva.
- `REQUIRED`: use the active transaction, or open a new one.
- `REQUIRES_NEW`: suspend the current transaction and open a new one.
- `SUPPORTS`: join the active transaction if there is one, otherwise run without a transaction.
- `NOT_SUPPORTED`: suspend the current transaction and run without one.
- `MANDATORY`: require an active transaction.
## Uso di `Repository`
## Synchronizations (`TxSynchronization`)
`Repository` è la base comune per le repository concrete.
Ogni operazione pubblica usa internamente una tx `REQUIRED` o `REQUIRED` read-only.
Lifecycle hooks for **one** transaction, registered through `Data.afterCommit(...)` (or directly
with `ResourceRegistry.addSynchronization(...)`).
Le sottoclassi implementano i metodi `doXxx(...)` del nuovo modello:
Every callback belongs to exactly the innermost transaction active at registration time, and fires
exactly once, when *that* transaction completes:
- a **joined** inner transaction (`REQUIRED`) is not a transaction of its own, so callbacks
registered inside one wait for the outermost commit;
- a `REQUIRES_NEW` transaction is, so completing it fires only its own callbacks and leaves the
suspended outer transaction's pending.
### Which side of the commit each hook sits on
| hook | when | resource |
| --- | --- | --- |
| `beforeCommit(readOnly)` | immediately **before** the real commit | session/connection still **bound**, transaction still active |
| `afterCommit()` / `afterRollback()` | after completion | resource already **unbound** |
| `afterCompletion(outcome)` | after the two above | resource already unbound |
`beforeCommit` is the only hook that can still write through the same resource and have the write
land in the same atomic unit: flush a buffer, stamp an audit row, materialize a derived value. It
is skipped when the transaction is already `rollback-only`, since there is no commit to precede.
Post-completion callbacks run with the resource unbound instead: one that opens its own transaction
gets a **fresh** one rather than joining the transaction that just finished. That is what makes
them the right place to refresh a cache, enqueue a message, or notify anything outside the
database.
### Failure
Throwing from `beforeCommit` **vetoes the commit**: the transaction is rolled back,
`afterRollback`/`afterCompletion(ROLLED_BACK)` fire, and the exception reaches the caller. That is
the reason the hook runs before the commit rather than after — it can still refuse.
The post-completion hooks have no such power: the transaction is already over by the time they run,
so an exception propagates but changes nothing already committed, and stops the callbacks queued
behind it.
## Using `Repository`
`Repository` is the shared base for concrete repositories. Every public operation internally uses a
`REQUIRED` transaction, read-only where applicable.
Subclasses implement the `doXxx(...)` methods:
- `doFind(Query<T>)`
- `doFindOne(Spec<T>)`
@@ -55,7 +95,8 @@ Le sottoclassi implementano i metodi `doXxx(...)` del nuovo modello:
- `doDeleteAll(Spec<T>)`
- `doUpdateAll(Spec<T>, T)`
I vecchi overload di `findAll(...)` e `findPage(...)` sono stati ridotti a una combinazione di `Query<T>` e `Spec<T>`.
The old `findAll(...)` and `findPage(...)` overloads were reduced to a combination of `Query<T>` and
`Spec<T>`.
```java
public abstract class Repository<T, ID> {
@@ -64,18 +105,21 @@ public abstract class Repository<T, ID> {
}
```
## Composizione con Flash
## Composing with Flash
`DataExtension` registra:
`DataExtension` registers:
- `Tx` nel `FlashContext`
- `TxManager` nel `FlashContext`
- un annotation processor per `@Transactional`
- `Tx` in the `FlashContext`
- `TxManager` in the `FlashContext`
- an annotation processor for `@Transactional`
Questo rende il layer dati componibile con il sistema di extension di Flash senza stato globale.
This makes the data layer composable with Flash's extension system without global state.
## Note implementative
## Implementation notes
- Lo stack transazionale è thread-local e viene ripulito quando torna vuoto.
- Le risorse backend sono sospese e ripristinate per `REQUIRES_NEW` e `NOT_SUPPORTED`.
- `TxSynchronization` è il punto di aggancio per hook di commit/rollback/completion.
- The transaction stack is thread-local and is cleared once it becomes empty.
- Backend resources are suspended and restored for `REQUIRES_NEW` and `NOT_SUPPORTED`.
- `TxSynchronization` is the hook point for commit/rollback/completion callbacks.
- Synchronizations live in a thread-local list; every new transaction records how many were already
registered when it opened and fires only its own tail, so a `REQUIRES_NEW` does not drag along
the suspended transaction's callbacks.
@@ -53,9 +53,52 @@ 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();
}
/**
* Runs {@link TxSynchronization#beforeCommit} on the synchronizations registered from
* {@code fromIndex} onward, while their transaction is still active and its resource still
* bound. Unlike {@link #fireSynchronizations} this leaves them registered: they still have
* their post-completion callbacks to come. Exceptions propagate on purpose — a beforeCommit
* that throws vetoes the commit, see {@link TxSynchronization}.
*
* <p>Snapshots before iterating, so a callback that registers further synchronizations (a
* nested {@code Data#afterCommit}) doesn't mutate the list mid-loop. Those new ones join the
* transaction's post-completion callbacks without getting a {@code beforeCommit} of their own,
* which is the only coherent answer once the pass is already running.
*/
public static void fireBeforeCommit(boolean readOnly, int fromIndex) {
List<TxSynchronization> pending = SYNCHRONIZATIONS.get();
if (fromIndex >= pending.size()) {
return;
}
for (TxSynchronization sync : List.copyOf(pending.subList(fromIndex, pending.size()))) {
sync.beforeCommit(readOnly);
}
}
/**
* 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();
@@ -1,8 +1,58 @@
package dev.relism.flash.ext.data.core;
/**
* Lifecycle hooks for one transaction, registered through {@code Data#afterCommit} (or
* {@link ResourceRegistry#addSynchronization} directly) and fired by the {@link TxManager} that
* owns the transaction they were registered in.
*
* <p>Every callback belongs to exactly one transaction — the innermost one active at registration
* time — and fires exactly once, when <em>that</em> transaction completes. A joined
* ({@code REQUIRED}) inner transaction is not a transaction of its own, so callbacks registered
* inside one wait for the outermost commit; a {@code REQUIRES_NEW} transaction is, so completing
* it fires only its own callbacks and leaves the suspended outer transaction's alone.
*
* <h3>Where each hook sits relative to the commit</h3>
* <ul>
* <li>{@link #beforeCommit(boolean)} — immediately <b>before</b> the real commit, with the
* transaction still active and its session/connection still bound. This is the only hook
* that can still write through that same resource and have the write land in the same atomic
* unit: flush a buffer, stamp an audit row, materialize a derived value. Skipped entirely
* when the transaction is already rollback-only, since there is no commit to precede.</li>
* <li>{@link #afterCommit()} / {@link #afterRollback()}, then {@link #afterCompletion(TxOutcome)}
* — <b>after</b> the transaction has completed and its resource has been unbound. Nothing
* done here is part of the transaction: a callback that opens its own transaction gets a
* fresh one instead of joining the one that just finished, which is what makes this the
* right place to refresh a cache, enqueue a message, or notify anything outside the
* database.</li>
* </ul>
*
* <h3>Failure</h3>
* Throwing from {@link #beforeCommit(boolean)} <b>vetoes the commit</b>: the transaction is rolled
* back, {@link #afterRollback()}/{@link #afterCompletion(TxOutcome)} fire with
* {@link TxOutcome#ROLLED_BACK}, and the exception propagates to the caller. That is the point of
* this hook running before the commit rather than after — it can still refuse.
*
* <p>The post-completion hooks have no such power: the transaction is over by the time they run,
* so an exception from one propagates to the caller but changes nothing already committed, and
* stops the callbacks queued behind it.
*/
public interface TxSynchronization {
/**
* Runs inside the transaction, immediately before it commits — see the interface javadoc.
* Throwing from here rolls the transaction back instead of committing it.
*
* @param readOnly whether the transaction was opened read-only, so a callback that would
* otherwise write can skip work it is not allowed to do
*/
default void beforeCommit(boolean readOnly) {}
/** Runs after a successful commit, with the transaction's resource already unbound. */
default void afterCommit() {}
/** Runs after a rollback, with the transaction's resource already unbound. */
default void afterRollback() {}
/** Runs after {@link #afterCommit()}/{@link #afterRollback()}, whichever applied. */
default void afterCompletion(TxOutcome outcome) {}
}
@@ -1,14 +1,15 @@
# flash-ext-data-hibernate
Backend Hibernate per `flash-ext-data-core`.
Hibernate backend for `flash-ext-data-core`.
## Scopo
## Purpose
Questo modulo implementa `TxManager` sopra `SessionFactory` e fornisce una base repository Hibernate-centric.
This module implements `TxManager` on top of a `SessionFactory` and provides a Hibernate-centric
repository base class.
## Come si usa
## How to use it
### 1. Creare il manager
### 1. Create the manager
```java
SessionFactory sessionFactory = ...;
@@ -16,12 +17,12 @@ HibernateTxManager txManager = new HibernateTxManager(sessionFactory);
DataExtension extension = new DataExtension(txManager);
```
### 2. Installare lestensione in Flash
### 2. Install the extension in Flash
Lestensione registra `Tx` e `TxManager` nel `FlashContext`.
Le handler class-based annotate con `@Transactional` vengono wrappate automaticamente.
The extension registers `Tx` and `TxManager` in the `FlashContext`. Class-based handlers annotated
with `@Transactional` are wrapped automatically.
### 3. Definire una repository
### 3. Define a repository
```java
public final class UserRepository extends HibernateRepository<User, Long> {
@@ -31,7 +32,7 @@ public final class UserRepository extends HibernateRepository<User, Long> {
}
```
Con il nuovo modello query/spec puoi esporre campi riusabili come costanti:
With the query/spec model you can expose reusable fields as constants:
```java
public final class UserRepository extends HibernateRepository<User, Long> {
@@ -48,7 +49,7 @@ public final class UserRepository extends HibernateRepository<User, Long> {
}
```
Le query domain-specific possono usare gli helper della base class:
Domain-specific queries can use the base class helpers:
```java
public List<User> findByEmailDomain(String domain) {
@@ -58,35 +59,37 @@ public List<User> findByEmailDomain(String domain) {
}
```
## Come funziona sotto
## How it works underneath
- La tx corrente è rappresentata da `HibernateTxStatus`.
- La risorsa esposta al core è una `Session`.
- `Tx.resource(Session.class)` recupera la `Session` dal contesto corrente.
- `REQUIRES_NEW` sospende lo status attivo e apre una nuova `Session`.
- `NOT_SUPPORTED` sospende la tx attiva e continua senza sessione bindata.
- The current transaction is represented by `HibernateTxStatus`.
- The resource exposed to the core is a `Session`.
- `Tx.resource(Session.class)` retrieves the `Session` from the current context.
- `REQUIRES_NEW` suspends the active status and opens a new `Session`.
- `NOT_SUPPORTED` suspends the active transaction and continues with no session bound.
## Repository base
## Repository base class
`HibernateRepository` fornisce:
`HibernateRepository` provides:
- `findById`, `findAll`, `findPage`, `findOne`
- `save`, `update`, `delete`, `saveAll`
- bulk `deleteAll(Spec<T>)` e `updateAll(Spec<T>, T)`
- helper HQL: `hql(...)`, `hqlMutate(...)`
- bulk `deleteAll(Spec<T>)` and `updateAll(Spec<T>, T)`
- HQL helpers: `hql(...)`, `hqlMutate(...)`
Le classi concrete devono solo implementare query di dominio, non il plumbing transazionale.
Concrete classes only have to implement domain queries, never the transactional plumbing.
## Semantica transazionale
## Transactional semantics
- `REQUIRED`: join o apertura nuova tx.
- `REQUIRES_NEW`: sospensione del contesto corrente.
- `SUPPORTS`: join se c’è tx, altrimenti no-op.
- `NOT_SUPPORTED`: sospende e prosegue senza tx.
- `MANDATORY`: fallisce se non c’è tx.
- `REQUIRED`: join, or open a new transaction.
- `REQUIRES_NEW`: suspend the current context.
- `SUPPORTS`: join if a transaction exists, otherwise no-op.
- `NOT_SUPPORTED`: suspend and continue without a transaction.
- `MANDATORY`: fail if there is no transaction.
## Note
## Notes
- `Session` viene chiusa a fine tx nuova.
- Le synchronizations vengono eseguite al commit/rollback.
- Il backend è pensato per essere usato tramite la base class, non direttamente.
- The `Session` is closed when a new transaction ends.
- Synchronizations registered in a transaction fire when *that* transaction completes:
`beforeCommit` while it is still active and the `Session` still bound, the post-completion hooks
once it is unbound. See `flash-ext-data-core/docs/README.md` for the full contract.
- This backend is meant to be used through the base class, not directly.
@@ -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,20 +121,45 @@ public class HibernateTxManager implements TxManager {
cleanupIfIdle();
return;
}
TxOutcome outcome;
TxOutcome outcome = null;
try {
if (s.isRollbackOnly() && s.session().getTransaction().isActive()) {
s.session().getTransaction().rollback();
outcome = TxOutcome.ROLLED_BACK;
} else {
// Still inside the transaction, session still bound: a beforeCommit callback can
// write through it and land in this same commit. Throwing from there vetoes the
// commit — see TxSynchronization.
ResourceRegistry.fireBeforeCommit(s.isReadOnly(), s.synchronizationBaseline());
s.session().getTransaction().commit();
outcome = TxOutcome.COMMITTED;
}
} catch (RuntimeException e) {
// A vetoing beforeCommit, or a commit that failed outright: either way nothing was
// committed, so roll back and let the remaining callbacks hear ROLLED_BACK rather than
// nothing at all. A rollback failure here is swallowed deliberately — it would mask
// the exception that actually explains what went wrong, which is the one propagating.
if (s.session().getTransaction().isActive()) {
try {
s.session().getTransaction().rollback();
} catch (RuntimeException suppressed) {
e.addSuppressed(suppressed);
}
}
outcome = TxOutcome.ROLLED_BACK;
throw e;
} 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 +175,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();
}
}
@@ -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; }
}
@@ -0,0 +1,258 @@
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 beforeCommit}/{@code afterCommit}/
* {@code afterRollback} callbacks {@code Data#afterCommit} exposes, and the contract callers build
* on: "my callback runs once, for my transaction, on the right side of the commit".
*
* <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 beforeCommit(boolean readOnly) { calls.add("beforeCommit:" + readOnly); }
@Override public void afterCommit() { calls.add("afterCommit"); }
@Override public void afterRollback() { calls.add("afterRollback"); }
@Override public void afterCompletion(TxOutcome outcome) { calls.add("afterCompletion:" + outcome); }
}
private static final List<String> COMMITTED = List.of("beforeCommit:false", "afterCommit", "afterCompletion:COMMITTED");
private static final List<String> ROLLED_BACK = List.of("afterRollback", "afterCompletion:ROLLED_BACK");
@Test
void afterCommit_fires_on_commit() {
Recorder recorder = new Recorder();
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
ResourceRegistry.addSynchronization(recorder);
manager.commit(tx);
assertEquals(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(ROLLED_BACK, recorder.calls);
}
/** A commit() call on a tx already marked rollback-only really rolls back — and has no commit to precede. */
@Test
void commit_of_a_rollback_only_tx_rolls_back_without_running_beforeCommit() {
Recorder recorder = new Recorder();
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
tx.markRollbackOnly();
ResourceRegistry.addSynchronization(recorder);
manager.commit(tx);
assertEquals(ROLLED_BACK, recorder.calls);
}
/** beforeCommit runs inside the transaction: session still bound, transaction still active. */
@Test
void beforeCommit_runs_while_the_transaction_is_still_active() {
List<Boolean> stillActive = new ArrayList<>();
List<Session> sessionSeen = new ArrayList<>();
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
Session session = tx.resource(Session.class);
ResourceRegistry.addSynchronization(new TxSynchronization() {
@Override
public void beforeCommit(boolean readOnly) {
stillActive.add(session.getTransaction().isActive());
// MANDATORY only succeeds while a transaction is bound — proof this runs inside it.
TxStatus joined = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.MANDATORY));
sessionSeen.add(joined.resource(Session.class));
manager.commit(joined);
}
});
manager.commit(tx);
assertEquals(List.of(true), stillActive, "the transaction must not have committed yet");
assertSame(session, sessionSeen.get(0), "the same session must still be bound, so writes land in this commit");
}
@Test
void beforeCommit_is_told_whether_the_transaction_is_read_only() {
Recorder readWrite = new Recorder();
TxStatus rw = manager.begin(TxDefinition.DEFAULTS);
ResourceRegistry.addSynchronization(readWrite);
manager.commit(rw);
Recorder readOnly = new Recorder();
TxStatus ro = manager.begin(TxDefinition.DEFAULTS.asReadOnly());
ResourceRegistry.addSynchronization(readOnly);
manager.commit(ro);
assertEquals("beforeCommit:false", readWrite.calls.get(0));
assertEquals("beforeCommit:true", readOnly.calls.get(0));
}
/** Throwing from beforeCommit is a veto: no commit, the rollback callbacks run, the exception propagates. */
@Test
void a_throwing_beforeCommit_vetoes_the_commit() {
Recorder recorder = new Recorder();
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
Session session = tx.resource(Session.class);
ResourceRegistry.addSynchronization(new TxSynchronization() {
@Override public void beforeCommit(boolean readOnly) { throw new IllegalStateException("veto"); }
});
ResourceRegistry.addSynchronization(recorder);
IllegalStateException thrown = assertThrows(IllegalStateException.class, () -> manager.commit(tx));
assertEquals("veto", thrown.getMessage());
assertFalse(session.getTransaction().isActive(), "the vetoed transaction must be rolled back, not left open");
assertEquals(ROLLED_BACK, recorder.calls, "the surviving callbacks must hear ROLLED_BACK, not silence");
}
/**
* The load-bearing ordering detail: post-completion 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(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(COMMITTED, innerSync.calls);
assertEquals(List.of(), outerSync.calls, "the outer transaction has not committed yet");
manager.commit(outer);
assertEquals(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(COMMITTED, outerSync.calls);
}
}
@@ -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);
}
}
@@ -1,14 +1,15 @@
# flash-ext-data-jdbc
Backend JDBC per `flash-ext-data-core`.
JDBC backend for `flash-ext-data-core`.
## Scopo
## Purpose
Questo modulo implementa `TxManager` sopra `DataSource` e fornisce una base repository SQL raw.
This module implements `TxManager` on top of a `DataSource` and provides a raw-SQL repository base
class.
## Come si usa
## How to use it
### 1. Creare il manager
### 1. Create the manager
```java
DataSource dataSource = ...;
@@ -16,11 +17,12 @@ JdbcTxManager txManager = new JdbcTxManager(dataSource);
DataExtension extension = new DataExtension(txManager);
```
### 2. Installare lestensione in Flash
### 2. Install the extension in Flash
Come per Hibernate, `DataExtension` registra `Tx` nel `FlashContext` e abilita `@Transactional` sugli handler class-based.
As with Hibernate, `DataExtension` registers `Tx` in the `FlashContext` and enables
`@Transactional` on class-based handlers.
### 3. Definire una repository
### 3. Define a repository
```java
public final class UserRepository extends JdbcRepository<User, Long> {
@@ -35,7 +37,7 @@ public final class UserRepository extends JdbcRepository<User, Long> {
}
```
Anche qui puoi esporre `Spec` riusabili e comporre query dal service layer:
Here too you can expose reusable `Spec`s and compose queries from the service layer:
```java
public final class UserRepository extends JdbcRepository<User, Long> {
@@ -47,7 +49,7 @@ public final class UserRepository extends JdbcRepository<User, Long> {
}
```
Per il salvataggio e lupdate devi fornire il binding esplicito:
Saving and updating need an explicit binding:
```java
@Override
@@ -61,36 +63,39 @@ protected void bindInsert(PreparedStatement ps, User entity) throws SQLException
}
```
## Come funziona sotto
## How it works underneath
- La tx corrente espone una `Connection`.
- `Tx.resource(Connection.class)` recupera la connessione bindata al thread.
- `REQUIRES_NEW` sospende la connessione attiva e ne apre una nuova.
- `NOT_SUPPORTED` sospende il contesto e prosegue senza tx.
- The current transaction exposes a `Connection`.
- `Tx.resource(Connection.class)` retrieves the connection bound to the thread.
- `REQUIRES_NEW` suspends the active connection and opens a new one.
- `NOT_SUPPORTED` suspends the context and continues without a transaction.
## Repository base
## Repository base class
`JdbcRepository` fornisce:
`JdbcRepository` provides:
- query `select` con `queryOne`, `queryMany`
- mutation con `mutate`
- persistenza con `doSave`, `doUpdate`
- paging con `doFindPage`
- `select` queries through `queryOne`, `queryMany`
- mutations through `mutate`
- persistence through `doSave`, `doUpdate`
- paging through `doFindPage`
- bulk `deleteAll(Spec<T>)`
- helper raw `queryOne(...)`, `queryMany(...)`, `mutate(...)`
- raw helpers `queryOne(...)`, `queryMany(...)`, `mutate(...)`
Le repository concrete devono solo tradurre tra `ResultSet` e dominio.
Concrete repositories only have to translate between `ResultSet` and the domain.
## Semantica transazionale
## Transactional semantics
- `REQUIRED`: join o apertura nuova tx.
- `REQUIRES_NEW`: sospensione del contesto corrente.
- `SUPPORTS`: join se c’è tx, altrimenti no-op.
- `NOT_SUPPORTED`: sospende e prosegue senza tx.
- `MANDATORY`: fallisce se non c’è tx.
- `REQUIRED`: join, or open a new transaction.
- `REQUIRES_NEW`: suspend the current context.
- `SUPPORTS`: join if a transaction exists, otherwise no-op.
- `NOT_SUPPORTED`: suspend and continue without a transaction.
- `MANDATORY`: fail if there is no transaction.
## Note
## Notes
- La `Connection` viene chiusa a fine tx nuova.
- Le synchronizations vengono eseguite al commit/rollback.
- Se una repository usa `doDelete(T)`, il comportamento predefinito è non supportato: usare `deleteById` o override specifico.
- The `Connection` is closed when a new transaction ends.
- Synchronizations registered in a transaction fire when *that* transaction completes:
`beforeCommit` while it is still active and the `Connection` still bound, the post-completion
hooks once it is unbound. See `flash-ext-data-core/docs/README.md` for the full contract.
- If a repository uses `doDelete(T)`, the default behaviour is unsupported: use `deleteById` or
override it.
@@ -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,22 +117,51 @@ 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 {
// Still inside the transaction, connection still bound: a beforeCommit callback
// can write through it and land in this same commit. Throwing from there vetoes
// the commit — see TxSynchronization.
ResourceRegistry.fireBeforeCommit(s.isReadOnly(), s.synchronizationBaseline());
s.connection().commit();
ResourceRegistry.fireSynchronizations(TxOutcome.COMMITTED);
outcome = TxOutcome.COMMITTED;
}
} catch (SQLException e) {
throw new TxException(e);
TxException wrapped = new TxException(e);
outcome = rollbackAfterFailedCommit(s, wrapped);
throw wrapped;
} catch (RuntimeException e) {
outcome = rollbackAfterFailedCommit(s, e);
throw 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();
}
}
/**
* Nothing was committed — a vetoing {@code beforeCommit}, or a commit that failed outright —
* so undo whatever the transaction had done and report {@code ROLLED_BACK} to the callbacks
* still queued behind it. A failure to roll back is attached to the exception already on its
* way out rather than replacing it: that one explains what actually went wrong.
*/
private static TxOutcome rollbackAfterFailedCommit(JdbcTxStatus s, Throwable propagating) {
try {
s.connection().rollback();
} catch (SQLException suppressed) {
propagating.addSuppressed(suppressed);
}
return TxOutcome.ROLLED_BACK;
}
@Override
public void rollback(TxStatus status) {
JdbcTxStatus s = (JdbcTxStatus) status;
@@ -137,11 +173,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();
}
}
@@ -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; }
}
@@ -0,0 +1,216 @@
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 beforeCommit(boolean readOnly) { calls.add("beforeCommit:" + readOnly); }
@Override public void afterCommit() { calls.add("afterCommit"); }
@Override public void afterRollback() { calls.add("afterRollback"); }
@Override public void afterCompletion(TxOutcome outcome) { calls.add("afterCompletion:" + outcome); }
}
private static final List<String> COMMITTED = List.of("beforeCommit:false", "afterCommit", "afterCompletion:COMMITTED");
private static final List<String> ROLLED_BACK = List.of("afterRollback", "afterCompletion:ROLLED_BACK");
@Test
void afterCommit_fires_on_commit() {
Recorder recorder = new Recorder();
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
ResourceRegistry.addSynchronization(recorder);
manager.commit(tx);
assertEquals(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(ROLLED_BACK, recorder.calls);
}
@Test
void commit_of_a_rollback_only_tx_rolls_back_without_running_beforeCommit() {
Recorder recorder = new Recorder();
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
tx.markRollbackOnly();
ResourceRegistry.addSynchronization(recorder);
manager.commit(tx);
assertEquals(ROLLED_BACK, recorder.calls);
}
/** beforeCommit runs inside the transaction: connection still bound, nothing committed yet. */
@Test
void beforeCommit_runs_while_the_transaction_is_still_active() {
List<Connection> connectionSeen = new ArrayList<>();
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
Connection connection = tx.resource(Connection.class);
ResourceRegistry.addSynchronization(new TxSynchronization() {
@Override
public void beforeCommit(boolean readOnly) {
// MANDATORY only succeeds while a transaction is bound — proof this runs inside it.
TxStatus joined = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.MANDATORY));
connectionSeen.add(joined.resource(Connection.class));
manager.commit(joined);
}
});
manager.commit(tx);
assertSame(connection, connectionSeen.get(0), "the same connection must still be bound, so writes land in this commit");
}
@Test
void beforeCommit_is_told_whether_the_transaction_is_read_only() {
Recorder readWrite = new Recorder();
TxStatus rw = manager.begin(TxDefinition.DEFAULTS);
ResourceRegistry.addSynchronization(readWrite);
manager.commit(rw);
Recorder readOnly = new Recorder();
TxStatus ro = manager.begin(TxDefinition.DEFAULTS.asReadOnly());
ResourceRegistry.addSynchronization(readOnly);
manager.commit(ro);
assertEquals("beforeCommit:false", readWrite.calls.get(0));
assertEquals("beforeCommit:true", readOnly.calls.get(0));
}
/** Throwing from beforeCommit is a veto: no commit, the rollback callbacks run, the exception propagates. */
@Test
void a_throwing_beforeCommit_vetoes_the_commit() {
Recorder recorder = new Recorder();
TxStatus tx = manager.begin(TxDefinition.DEFAULTS);
ResourceRegistry.addSynchronization(new TxSynchronization() {
@Override public void beforeCommit(boolean readOnly) { throw new IllegalStateException("veto"); }
});
ResourceRegistry.addSynchronization(recorder);
IllegalStateException thrown = assertThrows(IllegalStateException.class, () -> manager.commit(tx));
assertEquals("veto", thrown.getMessage());
assertEquals(ROLLED_BACK, recorder.calls, "the surviving callbacks must hear ROLLED_BACK, not silence");
}
/** Post-completion 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(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(COMMITTED, innerSync.calls);
assertEquals(List.of(), outerSync.calls, "the outer transaction has not committed yet");
manager.commit(outer);
assertEquals(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(COMMITTED, outerSync.calls);
}
}
@@ -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)));
}
}
@@ -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(); }
}