Initial pass at transaction support.

Now to test it!

pom.xml

@@ -46,7 +46,7 @@
     <dependency>
       <groupId>com.samskivert</groupId>
       <artifactId>samskivert</artifactId>
-      <version>1.8.2</version>
+      <version>1.9-SNAPSHOT</version>
     </dependency>
     <dependency>
       <groupId>com.google.guava</groupId>

src/main/java/com/samskivert/depot/PersistenceContext.java

@@ -585,10 +585,13 @@ public class PersistenceContext
         checkAreInitialized(); // le check du sanity
 
         boolean isReadOnly = op.isReadOnly();
-        Connection conn;
+        Transaction tx = Transaction.get();
+        ConnOp connop = (tx == null) ? new NonTxOp(isReadOnly) : new TxOp(tx);
+
         long preConnect = System.nanoTime();
+        Connection conn;
         try {
-            conn = _conprov.getConnection(_ident, isReadOnly);
+            conn = connop.get();
         } catch (PersistenceException pe) {
             throw new DatabaseException("Failed get connection [ident=" + _ident +
                                         ", isRO=" + isReadOnly + "]", pe);
@@ -616,12 +619,8 @@ public class PersistenceContext
                         stmt.close();
                     }
                 }
-                // Always commit if auto-commit is off.  If the read-only operation managed to
+                // let our connop do auto-commit, if appropriate
-                // acquire some locks, this will release them.  Also, we've seen a MySQL bug where
+                connop.done(conn);
-                // not committing after a select causes later selects to see stale results.
-                if (!conn.getAutoCommit()) {
-                    conn.commit();
-                }
                 // note the time it took to invoke this operation
                 _stats.noteOp(isReadOnly, preConnect, preInvoke, System.nanoTime());
                 // have the operation update any appropriate runtime statistics as well
@@ -637,10 +636,11 @@ public class PersistenceContext
                 }
 
                 // let the provider know that the connection failed
-                _conprov.connectionFailed(_ident, isReadOnly, conn, sqe);
+                boolean opAllowsRetry = connop.fail(conn, sqe);
                 conn = null;
 
-                if (retryOnTransientFailure && _liaison.isTransientException(sqe)) {
+                if (retryOnTransientFailure && opAllowsRetry &&
+                    _liaison.isTransientException(sqe)) {
                     // the MySQL JDBC driver has the annoying habit of including the embedded
                     // exception stack trace in the message of their outer exception; if I want a
                     // fucking stack trace, I'll call printStackTrace() thanksverymuch
@@ -653,7 +653,7 @@ public class PersistenceContext
 
             } finally {
                 if (conn != null) {
-                    _conprov.releaseConnection(_ident, isReadOnly, conn);
+                    connop.release(conn);
                 }
             }
         }
@@ -671,6 +671,58 @@ public class PersistenceContext
         }
     }
 
+    interface ConnOp {
+        Connection get () throws PersistenceException;
+        void done (Connection conn) throws SQLException;
+        boolean fail (Connection conn, SQLException sqe);
+        void release (Connection conn);
+    }
+
+    protected class NonTxOp implements ConnOp {
+        public final boolean readOnly;
+        public NonTxOp (boolean readOnly) {
+            this.readOnly = readOnly;
+        }
+        public Connection get () throws PersistenceException {
+            return _conprov.getConnection(_ident, readOnly);
+        }
+        public void done (Connection conn) throws SQLException {
+            // Always commit if auto-commit is off. If the read-only operation managed to acquire
+            // some locks, this will release them. Also, we've seen a MySQL bug where not
+            // committing after a select causes later selects to see stale results.
+            if (!conn.getAutoCommit()) {
+                conn.commit();
+            }
+        }
+        public boolean fail (Connection conn, SQLException sqe) {
+            _conprov.connectionFailed(_ident, readOnly, conn, sqe);
+            return true;
+        }
+        public void release (Connection conn) {
+            _conprov.releaseConnection(_ident, readOnly, conn);
+        }
+    };
+
+    protected class TxOp implements ConnOp {
+        public final Transaction tx;
+        public TxOp (Transaction tx) {
+            this.tx = tx;
+        }
+        public Connection get () throws PersistenceException {
+            return tx.getConnection();
+        }
+        public void done (Connection conn) throws SQLException {
+            // nothing to do here, we commit (or not) when the whole tx is done
+        }
+        public boolean fail (Connection conn, SQLException sqe) {
+            tx.connectionFailed(sqe);
+            return false;
+        }
+        public void release (Connection conn) {
+            // nothing to do here, we release the conn when the whole tx is done
+        }
+    }
+
     protected final CanMigrate _canMigrate;
     protected String _ident;
     protected ConnectionProvider _conprov;

src/main/java/com/samskivert/depot/Transaction.java

@@ -0,0 +1,164 @@
+//
+// Depot library - a Java relational persistence library
+// https://github.com/threerings/depot/blob/master/LICENSE
+
+package com.samskivert.depot;
+
+import com.samskivert.io.PersistenceException;
+import com.samskivert.jdbc.ConnectionProvider;
+import java.sql.Connection;
+import java.sql.SQLException;
+
+/**
+ * Allows database operations to be performed in a transaction. Transactions can be performed
+ * manually as follows:
+ * <pre>{@code
+ * PersistenceContext ctx = ...;
+ * Transaction tx = Transaction.start(ctx);
+ * try {
+ *   _someRepo.updateFoo(...);
+ *   _someOtherRepo.updateBar(...);
+ *   tx.commit();
+ * } catch (RuntimeException re)  {
+ *   tx.rollback();
+ *   throw re;
+ * }
+ * }</pre>
+ *
+ * Or one can make use of the {@link Transaction#perform} helper method:
+ * <pre>{@code
+ * PersistenceContext ctx = ...;
+ * Transaction.perform(ctx, new Runnable() {
+ *   public void run () {
+ *     _someRepo.updateFoo(...);
+ *     _someOtherRepo.updateBar(...);
+ *   }
+ * });
+ * }</pre>
+ *
+ * <p><b>Caveats:</b></p>
+ *
+ * <p>All repositories involved in a transaction must use the same {@code PersistenceContext}
+ * supplied when starting the transaction. Attempting to use a repository which references a
+ * different context will cause an exception to be thrown (and the transaction to be aborted,
+ * assuming you've structured your code correctly).</p>
+ *
+ * <p>Transactions may not be nested.</p>
+ */
+public class Transaction {
+
+    /**
+     * Starts a transaction using the supplied persistence context. You <em>must</em> eventually
+     * call {@link #commit} or {@link #rollback} on the returned transaction.
+     */
+    public static Transaction start (PersistenceContext ctx)
+    {
+        Transaction tx = get();
+        if (tx != null) throw new DatabaseException("Nested transactions not supported.");
+        tx = new Transaction(ctx);
+        _activeTx.set(tx);
+        return tx;
+    }
+
+    /**
+     * Returns the currently active transaction (on the caller's thread), or null.
+     */
+    public static Transaction get ()
+    {
+        return _activeTx.get();
+    }
+
+    /**
+     * Returns true if a transaction is currently active on the caller's thread.
+     */
+    public static boolean inTransaction ()
+    {
+        return get() != null;
+    }
+
+    /**
+     * Performs {@code op} inside a transaction, committing it if {@code op} completes
+     * successfully, rolling it back if {@code op} throws any exceptions.
+     */
+    public static void perform (PersistenceContext ctx, Runnable op)
+    {
+        Transaction tx = start(ctx);
+        try {
+            op.run();
+            tx.commit();
+        } catch (RuntimeException re) {
+            tx.rollback();
+            throw re;
+        }
+    }
+
+    /**
+     * Commits this transaction.
+     */
+    public void commit ()
+    {
+        if (_conn != null) {
+            checkActive("commit");
+            try {
+                _conn.commit();
+            } catch (SQLException sqe) {
+                throw new DatabaseException("Transaction commit failure", sqe);
+            } finally {
+                release();
+            }
+        }
+    }
+
+    /**
+     * Aborts this transaction, rolling back any database operations performed thus far.
+     */
+    public void rollback ()
+    {
+        if (_conn != null) {
+            checkActive("rollback");
+            try {
+                _conn.rollback();
+            } catch (SQLException sqe) {
+                throw new DatabaseException("Transaction rollback failure", sqe);
+            } finally {
+                release();
+            }
+        }
+    }
+
+    /** The persistence context in which this transaction is operating. */
+    public final PersistenceContext ctx;
+
+    Connection getConnection ()
+        throws PersistenceException
+    {
+        if (_conn == null) _conn = ctx._conprov.getTxConnection(ctx._ident);
+        return _conn;
+    }
+
+    void connectionFailed (SQLException sqe)
+    {
+        ctx._conprov.txConnectionFailed(ctx._ident, _conn, sqe);
+    }
+
+    protected Transaction (PersistenceContext ctx)
+    {
+        this.ctx = ctx;
+    }
+
+    protected void checkActive (String action) {
+        if (_activeTx.get() != this) throw new IllegalStateException(
+            "Attempted to " + action + " non-active transaction");
+    }
+
+    protected void release () {
+        ctx._conprov.releaseTxConnection(ctx._ident, _conn);
+        _conn = null;
+        _activeTx.set(null);
+    }
+
+    protected static final ThreadLocal<Transaction> _activeTx = new ThreadLocal<Transaction>();
+
+    /** The connection being used for this transaction. */
+    protected Connection _conn;
+}