diff --git a/flash-extensions/flash-ext-data-core/docs/README.md b/flash-extensions/flash-ext-data-core/docs/README.md index 8e19234..df53379 100644 --- a/flash-extensions/flash-ext-data-core/docs/README.md +++ b/flash-extensions/flash-ext-data-core/docs/README.md @@ -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`: base repository auto-transazionale. -- `Spec`: predicato componibile. -- `Query`: oggetto query con spec, sort e paging. -- `SpecBuilder`: DSL fluente per costruire spec tipizzate. -- `RepositorySupport`: 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`: self-transactional base repository. +- `Spec`: composable predicate. +- `Query`: query object carrying spec, sort and paging. +- `SpecBuilder`: fluent DSL for building typed specs. +- `RepositorySupport`: 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)` - `doFindOne(Spec)` @@ -55,7 +95,8 @@ Le sottoclassi implementano i metodi `doXxx(...)` del nuovo modello: - `doDeleteAll(Spec)` - `doUpdateAll(Spec, T)` -I vecchi overload di `findAll(...)` e `findPage(...)` sono stati ridotti a una combinazione di `Query` e `Spec`. +The old `findAll(...)` and `findPage(...)` overloads were reduced to a combination of `Query` and +`Spec`. ```java public abstract class Repository { @@ -64,18 +105,21 @@ public abstract class Repository { } ``` -## 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. diff --git a/flash-extensions/flash-ext-data-core/src/main/java/dev/relism/flash/ext/data/core/ResourceRegistry.java b/flash-extensions/flash-ext-data-core/src/main/java/dev/relism/flash/ext/data/core/ResourceRegistry.java index 5b491f0..1af31ce 100644 --- a/flash-extensions/flash-ext-data-core/src/main/java/dev/relism/flash/ext/data/core/ResourceRegistry.java +++ b/flash-extensions/flash-ext-data-core/src/main/java/dev/relism/flash/ext/data/core/ResourceRegistry.java @@ -53,9 +53,52 @@ public final class ResourceRegistry { SYNCHRONIZATIONS.get().add(Objects.requireNonNull(sync)); } - public static void fireSynchronizations(TxOutcome outcome) { - List 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}. + * + *

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 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 suspended, 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 pending = SYNCHRONIZATIONS.get(); + if (fromIndex >= pending.size()) { + return; + } + List syncs = List.copyOf(pending.subList(fromIndex, pending.size())); + pending.subList(fromIndex, pending.size()).clear(); for (TxSynchronization sync : syncs) { if (outcome == TxOutcome.COMMITTED) { sync.afterCommit(); diff --git a/flash-extensions/flash-ext-data-core/src/main/java/dev/relism/flash/ext/data/core/TxSynchronization.java b/flash-extensions/flash-ext-data-core/src/main/java/dev/relism/flash/ext/data/core/TxSynchronization.java index 380d51a..57ce363 100644 --- a/flash-extensions/flash-ext-data-core/src/main/java/dev/relism/flash/ext/data/core/TxSynchronization.java +++ b/flash-extensions/flash-ext-data-core/src/main/java/dev/relism/flash/ext/data/core/TxSynchronization.java @@ -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. + * + *

Every callback belongs to exactly one transaction — the innermost one active at registration + * time — and fires exactly once, when that 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. + * + *

Where each hook sits relative to the commit

+ *
    + *
  • {@link #beforeCommit(boolean)} — immediately before 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.
    • + *
  • {@link #afterCommit()} / {@link #afterRollback()}, then {@link #afterCompletion(TxOutcome)} + * — after 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.
    • + *
+ * + *

Failure

+ * Throwing from {@link #beforeCommit(boolean)} vetoes the commit: 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. + * + *

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) {} } diff --git a/flash-extensions/flash-ext-data-hibernate/docs/README.md b/flash-extensions/flash-ext-data-hibernate/docs/README.md index 5264a02..5854b6a 100644 --- a/flash-extensions/flash-ext-data-hibernate/docs/README.md +++ b/flash-extensions/flash-ext-data-hibernate/docs/README.md @@ -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 l’estensione in Flash +### 2. Install the extension in Flash -L’estensione 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 { @@ -31,7 +32,7 @@ public final class UserRepository extends HibernateRepository { } ``` -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 { @@ -48,7 +49,7 @@ public final class UserRepository extends HibernateRepository { } ``` -Le query domain-specific possono usare gli helper della base class: +Domain-specific queries can use the base class helpers: ```java public List findByEmailDomain(String domain) { @@ -58,35 +59,37 @@ public List 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)` e `updateAll(Spec, T)` -- helper HQL: `hql(...)`, `hqlMutate(...)` +- bulk `deleteAll(Spec)` and `updateAll(Spec, 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. diff --git a/flash-extensions/flash-ext-data-hibernate/src/main/java/dev/relism/flash/ext/data/hibernate/HibernateTxManager.java b/flash-extensions/flash-ext-data-hibernate/src/main/java/dev/relism/flash/ext/data/hibernate/HibernateTxManager.java index 5ce0901..537893c 100644 --- a/flash-extensions/flash-ext-data-hibernate/src/main/java/dev/relism/flash/ext/data/hibernate/HibernateTxManager.java +++ b/flash-extensions/flash-ext-data-hibernate/src/main/java/dev/relism/flash/ext/data/hibernate/HibernateTxManager.java @@ -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(); } } diff --git a/flash-extensions/flash-ext-data-hibernate/src/main/java/dev/relism/flash/ext/data/hibernate/HibernateTxStatus.java b/flash-extensions/flash-ext-data-hibernate/src/main/java/dev/relism/flash/ext/data/hibernate/HibernateTxStatus.java index 5a6a3e6..0ecccfc 100644 --- a/flash-extensions/flash-ext-data-hibernate/src/main/java/dev/relism/flash/ext/data/hibernate/HibernateTxStatus.java +++ b/flash-extensions/flash-ext-data-hibernate/src/main/java/dev/relism/flash/ext/data/hibernate/HibernateTxStatus.java @@ -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; } } diff --git a/flash-extensions/flash-ext-data-hibernate/src/test/java/dev/relism/flash/ext/data/hibernate/HibernateTxManagerSynchronizationTest.java b/flash-extensions/flash-ext-data-hibernate/src/test/java/dev/relism/flash/ext/data/hibernate/HibernateTxManagerSynchronizationTest.java new file mode 100644 index 0000000..6288ebe --- /dev/null +++ b/flash-extensions/flash-ext-data-hibernate/src/test/java/dev/relism/flash/ext/data/hibernate/HibernateTxManagerSynchronizationTest.java @@ -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". + * + *

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 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 COMMITTED = List.of("beforeCommit:false", "afterCommit", "afterCompletion:COMMITTED"); + private static final List 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 stillActive = new ArrayList<>(); + List 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 after 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 sessionsSeen = new ArrayList<>(); + List 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); + } +} diff --git a/flash-extensions/flash-ext-data-hibernate/src/test/java/dev/relism/flash/ext/data/hibernate/HibernateTxManagerTest.java b/flash-extensions/flash-ext-data-hibernate/src/test/java/dev/relism/flash/ext/data/hibernate/HibernateTxManagerTest.java index 47552cc..e3f4d49 100644 --- a/flash-extensions/flash-ext-data-hibernate/src/test/java/dev/relism/flash/ext/data/hibernate/HibernateTxManagerTest.java +++ b/flash-extensions/flash-ext-data-hibernate/src/test/java/dev/relism/flash/ext/data/hibernate/HibernateTxManagerTest.java @@ -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); + } } diff --git a/flash-extensions/flash-ext-data-jdbc/docs/README.md b/flash-extensions/flash-ext-data-jdbc/docs/README.md index 62e9cd4..fd3b4cb 100644 --- a/flash-extensions/flash-ext-data-jdbc/docs/README.md +++ b/flash-extensions/flash-ext-data-jdbc/docs/README.md @@ -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 l’estensione 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 { @@ -35,7 +37,7 @@ public final class UserRepository extends JdbcRepository { } ``` -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 { @@ -47,7 +49,7 @@ public final class UserRepository extends JdbcRepository { } ``` -Per il salvataggio e l’update 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)` -- 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. diff --git a/flash-extensions/flash-ext-data-jdbc/src/main/java/dev/relism/flash/ext/data/jdbc/JdbcTxManager.java b/flash-extensions/flash-ext-data-jdbc/src/main/java/dev/relism/flash/ext/data/jdbc/JdbcTxManager.java index 9a87f8d..08ad9b1 100644 --- a/flash-extensions/flash-ext-data-jdbc/src/main/java/dev/relism/flash/ext/data/jdbc/JdbcTxManager.java +++ b/flash-extensions/flash-ext-data-jdbc/src/main/java/dev/relism/flash/ext/data/jdbc/JdbcTxManager.java @@ -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(); + outcome = TxOutcome.COMMITTED; } - s.connection().commit(); - ResourceRegistry.fireSynchronizations(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(); } } diff --git a/flash-extensions/flash-ext-data-jdbc/src/main/java/dev/relism/flash/ext/data/jdbc/JdbcTxStatus.java b/flash-extensions/flash-ext-data-jdbc/src/main/java/dev/relism/flash/ext/data/jdbc/JdbcTxStatus.java index cf7f45d..f05c981 100644 --- a/flash-extensions/flash-ext-data-jdbc/src/main/java/dev/relism/flash/ext/data/jdbc/JdbcTxStatus.java +++ b/flash-extensions/flash-ext-data-jdbc/src/main/java/dev/relism/flash/ext/data/jdbc/JdbcTxStatus.java @@ -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; } } diff --git a/flash-extensions/flash-ext-data-jdbc/src/test/java/dev/relism/flash/ext/data/jdbc/JdbcTxManagerSynchronizationTest.java b/flash-extensions/flash-ext-data-jdbc/src/test/java/dev/relism/flash/ext/data/jdbc/JdbcTxManagerSynchronizationTest.java new file mode 100644 index 0000000..8a6e83e --- /dev/null +++ b/flash-extensions/flash-ext-data-jdbc/src/test/java/dev/relism/flash/ext/data/jdbc/JdbcTxManagerSynchronizationTest.java @@ -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 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 COMMITTED = List.of("beforeCommit:false", "afterCommit", "afterCompletion:COMMITTED"); + private static final List 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 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 connectionsSeen = new ArrayList<>(); + List 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); + } +} diff --git a/flash-extensions/flash-ext-data-jdbc/src/test/java/dev/relism/flash/ext/data/jdbc/JdbcTxManagerTest.java b/flash-extensions/flash-ext-data-jdbc/src/test/java/dev/relism/flash/ext/data/jdbc/JdbcTxManagerTest.java index 8334569..50e30f5 100644 --- a/flash-extensions/flash-ext-data-jdbc/src/test/java/dev/relism/flash/ext/data/jdbc/JdbcTxManagerTest.java +++ b/flash-extensions/flash-ext-data-jdbc/src/test/java/dev/relism/flash/ext/data/jdbc/JdbcTxManagerTest.java @@ -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)); - @Override - public Connection getConnection(String username, String password) throws SQLException { - return DriverManager.getConnection("jdbc:h2:mem:tx-jdbc;DB_CLOSE_DELAY=-1", username, password); - } + assertFalse(s.isNewTransaction()); + assertThrows(IllegalStateException.class, () -> s.resource(Connection.class)); + assertDoesNotThrow(() -> manager.commit(s)); + } - @Override - public T unwrap(Class iface) { - throw new UnsupportedOperationException(); - } + @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); - @Override - public boolean isWrapperFor(Class iface) { - return false; - } + TxStatus suspended = manager.begin(TxDefinition.DEFAULTS.withPropagation(TransactionPropagation.NOT_SUPPORTED)); + assertFalse(suspended.isNewTransaction()); + assertThrows(IllegalStateException.class, () -> suspended.resource(Connection.class)); + manager.commit(suspended); - @Override - public java.io.PrintWriter getLogWriter() { - throw new UnsupportedOperationException(); - } + 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 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))); } } diff --git a/flash-extensions/flash-ext-data-jdbc/src/test/java/dev/relism/flash/ext/data/jdbc/TestDataSource.java b/flash-extensions/flash-ext-data-jdbc/src/test/java/dev/relism/flash/ext/data/jdbc/TestDataSource.java new file mode 100644 index 0000000..54c19c2 --- /dev/null +++ b/flash-extensions/flash-ext-data-jdbc/src/test/java/dev/relism/flash/ext/data/jdbc/TestDataSource.java @@ -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 unwrap(Class 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(); } +}