X-Git-Url: https://git.opendaylight.org/gerrit/gitweb?a=blobdiff_plain;ds=sidebyside;f=opendaylight%2Fmd-sal%2Fsal-distributed-datastore%2Fsrc%2Fmain%2Fjava%2Forg%2Fopendaylight%2Fcontroller%2Fcluster%2Fdatabroker%2Factors%2Fdds%2FAbstractProxyTransaction.java;h=84546632f099bb3302938f6d0e8990c3f93e5e3b;hb=da42d2ffc8904b8dd24596cf6d918a0d30c8c521;hp=36f9a4bccbe07e762dd8f4778ae4809fc4a2487a;hpb=b4d95acff78952020e9fbde4372d13b461fd7469;p=controller.git

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java
index 36f9a4bccb..84546632f0 100644
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/databroker/actors/dds/AbstractProxyTransaction.java
@@ -8,10 +8,12 @@
 package org.opendaylight.controller.cluster.databroker.actors.dds;
 
 import akka.actor.ActorRef;
+import com.google.common.base.MoreObjects;
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import com.google.common.base.Throwables;
 import com.google.common.base.Verify;
+import com.google.common.collect.Iterables;
 import com.google.common.util.concurrent.CheckedFuture;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
@@ -27,6 +29,9 @@ import javax.annotation.Nullable;
 import javax.annotation.concurrent.GuardedBy;
 import javax.annotation.concurrent.NotThreadSafe;
 import org.opendaylight.controller.cluster.access.client.ConnectionEntry;
+import org.opendaylight.controller.cluster.access.commands.AbstractLocalTransactionRequest;
+import org.opendaylight.controller.cluster.access.commands.ClosedTransactionException;
+import org.opendaylight.controller.cluster.access.commands.IncrementTransactionSequenceRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionAbortRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionAbortSuccess;
 import org.opendaylight.controller.cluster.access.commands.TransactionCanCommitSuccess;
@@ -34,6 +39,7 @@ import org.opendaylight.controller.cluster.access.commands.TransactionCommitSucc
 import org.opendaylight.controller.cluster.access.commands.TransactionDoCommitRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionPreCommitSuccess;
+import org.opendaylight.controller.cluster.access.commands.TransactionPurgeRequest;
 import org.opendaylight.controller.cluster.access.commands.TransactionRequest;
 import org.opendaylight.controller.cluster.access.concepts.Request;
 import org.opendaylight.controller.cluster.access.concepts.RequestFailure;
@@ -66,18 +72,30 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      */
     @NotThreadSafe
     private static final class IncrementSequence {
-        private long delta = 1;
+        private final long sequence;
+        private long delta = 0;
+
+        IncrementSequence(final long sequence) {
+            this.sequence = sequence;
+        }
 
         long getDelta() {
             return delta;
         }
 
+        long getSequence() {
+            return sequence;
+        }
+
         void incrementDelta() {
             delta++;
         }
     }
 
-    // Generic state base class. Direct instances are used for fast paths, sub-class is used for successor transitions
+    /**
+     * Base class for representing logical state of this proxy. See individual instantiations and {@link SuccessorState}
+     * for details.
+     */
     private static class State {
         private final String string;
 
@@ -91,14 +109,27 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
     }
 
-    // State class used when a successor has interfered. Contains coordinator latch, the successor and previous state
-    private static final class SuccessorState extends State {
+    /**
+     * State class used when a successor has interfered. Contains coordinator latch, the successor and previous state.
+     * This is a temporary state introduced during reconnection process and is necessary for correct state hand-off
+     * between the old connection (potentially being accessed by the user) and the new connection (being cleaned up
+     * by the actor.
+     *
+     * <p>
+     * When a user operation encounters this state, it synchronizes on the it and wait until reconnection completes,
+     * at which point the request is routed to the successor transaction. This is a relatively heavy-weight solution
+     * to the problem of state transfer, but the user will observe it only if the race condition is hit.
+     */
+    private static class SuccessorState extends State {
         private final CountDownLatch latch = new CountDownLatch(1);
         private AbstractProxyTransaction successor;
         private State prevState;
 
+        // SUCCESSOR + DONE
+        private boolean done;
+
         SuccessorState() {
-            super("successor");
+            super("SUCCESSOR");
         }
 
         // Synchronize with succession process and return the successor
@@ -117,12 +148,15 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
 
         State getPrevState() {
-            return prevState;
+            return Verify.verifyNotNull(prevState, "Attempted to access previous state, which was not set");
         }
 
         void setPrevState(final State prevState) {
-            Verify.verify(this.prevState == null);
+            Verify.verify(this.prevState == null, "Attempted to set previous state to %s when we already have %s",
+                    prevState, this.prevState);
             this.prevState = Preconditions.checkNotNull(prevState);
+            // We cannot have duplicate successor states, so this check is sufficient
+            this.done = DONE.equals(prevState);
         }
 
         // To be called from safe contexts, where successor is known to be completed
@@ -131,9 +165,18 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
 
         void setSuccessor(final AbstractProxyTransaction successor) {
-            Verify.verify(this.successor == null);
+            Verify.verify(this.successor == null, "Attempted to set successor to %s when we already have %s",
+                    successor, this.successor);
             this.successor = Preconditions.checkNotNull(successor);
         }
+
+        boolean isDone() {
+            return done;
+        }
+
+        void setDone() {
+            done = true;
+        }
     }
 
     private static final Logger LOG = LoggerFactory.getLogger(AbstractProxyTransaction.class);
@@ -141,9 +184,41 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             AtomicIntegerFieldUpdater.newUpdater(AbstractProxyTransaction.class, "sealed");
     private static final AtomicReferenceFieldUpdater<AbstractProxyTransaction, State> STATE_UPDATER =
             AtomicReferenceFieldUpdater.newUpdater(AbstractProxyTransaction.class, State.class, "state");
-    private static final State OPEN = new State("open");
-    private static final State SEALED = new State("sealed");
-    private static final State FLUSHED = new State("flushed");
+
+    /**
+     * Transaction has been open and is being actively worked on.
+     */
+    private static final State OPEN = new State("OPEN");
+
+    /**
+     * Transaction has been sealed by the user, but it has not completed flushing to the backed, yet. This is
+     * a transition state, as we are waiting for the user to initiate commit procedures.
+     *
+     * <p>
+     * Since the reconnect mechanics relies on state replay for transactions, this state needs to be flushed into the
+     * queue to re-create state in successor transaction (which may be based on different messages as locality may have
+     * changed). Hence the transition to {@link #FLUSHED} state needs to be handled in a thread-safe manner.
+     */
+    private static final State SEALED = new State("SEALED");
+
+    /**
+     * Transaction state has been flushed into the queue, i.e. it is visible by the successor and potentially
+     * the backend. At this point the transaction does not hold any state besides successful requests, all other state
+     * is held either in the connection's queue or the successor object.
+     *
+     * <p>
+     * Transition to this state indicates we have all input from the user we need to initiate the correct commit
+     * protocol.
+     */
+    private static final State FLUSHED = new State("FLUSHED");
+
+    /**
+     * Transaction state has been completely resolved, we have received confirmation of the transaction fate from
+     * the backend. The only remaining task left to do is finishing up the state cleanup, which is done via purge
+     * request. We need to hang on to the transaction until that is done, as we have to make sure backend completes
+     * purging its state -- otherwise we could have a leak on the backend.
+     */
+    private static final State DONE = new State("DONE");
 
     // Touched from client actor thread only
     private final Deque<Object> successfulRequests = new ArrayDeque<>();
@@ -176,18 +251,32 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * variable. It uses pre-allocated objects for fast paths (i.e. no successor present) and a per-transition object
      * for slow paths (when successor is injected/present).
      */
-    private volatile int sealed = 0;
-    private volatile State state = OPEN;
+    private volatile int sealed;
+    private volatile State state;
 
-    AbstractProxyTransaction(final ProxyHistory parent) {
+    AbstractProxyTransaction(final ProxyHistory parent, final boolean isDone) {
         this.parent = Preconditions.checkNotNull(parent);
+        if (isDone) {
+            state = DONE;
+            // DONE implies previous seal operation completed
+            sealed = 1;
+        } else {
+            state = OPEN;
+        }
+    }
+
+    final void executeInActor(final Runnable command) {
+        parent.context().executeInActor(behavior -> {
+            command.run();
+            return behavior;
+        });
     }
 
     final ActorRef localActor() {
         return parent.localActor();
     }
 
-    private void incrementSequence(final long delta) {
+    final void incrementSequence(final long delta) {
         sequence += delta;
         LOG.debug("Transaction {} incremented sequence to {}", this, sequence);
     }
@@ -199,16 +288,19 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     }
 
     final void delete(final YangInstanceIdentifier path) {
+        checkReadWrite();
         checkNotSealed();
         doDelete(path);
     }
 
     final void merge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
+        checkReadWrite();
         checkNotSealed();
         doMerge(path, data);
     }
 
     final void write(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data) {
+        checkReadWrite();
         checkNotSealed();
         doWrite(path, data);
     }
@@ -223,6 +315,12 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         return doRead(path);
     }
 
+    final void enqueueRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback,
+            final long enqueuedTicks) {
+        LOG.debug("Transaction proxy {} enqueing request {} callback {}", this, request, callback);
+        parent.enqueueRequest(request, callback, enqueuedTicks);
+    }
+
     final void sendRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
         LOG.debug("Transaction proxy {} sending request {} callback {}", this, request, callback);
         parent.sendRequest(request, callback);
@@ -235,6 +333,16 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         // Transition user-visible state first
         final boolean success = SEALED_UPDATER.compareAndSet(this, 0, 1);
         Preconditions.checkState(success, "Proxy %s was already sealed", getIdentifier());
+        internalSeal();
+    }
+
+    final void ensureSealed() {
+        if (SEALED_UPDATER.compareAndSet(this, 0, 1)) {
+            internalSeal();
+        }
+    }
+
+    private void internalSeal() {
         doSeal();
         parent.onTransactionSealed(this);
 
@@ -247,7 +355,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             // still stuck here. The successor has not seen final part of our state, nor the fact it is sealed.
             // Propagate state and seal the successor.
             flushState(successor);
-            successor.seal();
+            successor.ensureSealed();
         }
     }
 
@@ -265,16 +373,22 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         return (SuccessorState) local;
     }
 
+    private void checkReadWrite() {
+        if (isSnapshotOnly()) {
+            throw new UnsupportedOperationException("Transaction " + getIdentifier() + " is a read-only snapshot");
+        }
+    }
+
     final void recordSuccessfulRequest(final @Nonnull TransactionRequest<?> req) {
         successfulRequests.add(Verify.verifyNotNull(req));
     }
 
-    final void recordFinishedRequest() {
+    final void recordFinishedRequest(final Response<?, ?> response) {
         final Object last = successfulRequests.peekLast();
         if (last instanceof IncrementSequence) {
             ((IncrementSequence) last).incrementDelta();
         } else {
-            successfulRequests.addLast(new IncrementSequence());
+            successfulRequests.addLast(new IncrementSequence(response.getSequence()));
         }
     }
 
@@ -284,29 +398,51 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      */
     final void abort() {
         checkNotSealed();
-        doAbort();
         parent.abortTransaction(this);
+
+        sendRequest(abortRequest(), resp -> {
+            LOG.debug("Transaction {} abort completed with {}", getIdentifier(), resp);
+            enqueuePurge();
+        });
     }
 
     final void abort(final VotingFuture<Void> ret) {
         checkSealed();
 
-        sendAbort(t -> {
+        sendDoAbort(t -> {
             if (t instanceof TransactionAbortSuccess) {
                 ret.voteYes();
             } else if (t instanceof RequestFailure) {
-                ret.voteNo(((RequestFailure<?, ?>) t).getCause());
+                ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
             } else {
                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
             }
 
             // This is a terminal request, hence we do not need to record it
             LOG.debug("Transaction {} abort completed", this);
-            parent.completeTransaction(this);
+            enqueuePurge();
         });
     }
 
-    final void sendAbort(final Consumer<Response<?, ?>> callback) {
+    final void enqueueAbort(final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+        checkNotSealed();
+        parent.abortTransaction(this);
+
+        enqueueRequest(abortRequest(), resp -> {
+            LOG.debug("Transaction {} abort completed with {}", getIdentifier(), resp);
+            // Purge will be sent by the predecessor's callback
+            if (callback != null) {
+                callback.accept(resp);
+            }
+        }, enqueuedTicks);
+    }
+
+    final void enqueueDoAbort(final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+        enqueueRequest(new TransactionAbortRequest(getIdentifier(), nextSequence(), localActor()), callback,
+            enqueuedTicks);
+    }
+
+    final void sendDoAbort(final Consumer<Response<?, ?>> callback) {
         sendRequest(new TransactionAbortRequest(getIdentifier(), nextSequence(), localActor()), callback);
     }
 
@@ -317,6 +453,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * @return Future completion
      */
     final ListenableFuture<Boolean> directCommit() {
+        checkReadWrite();
         checkSealed();
 
         // Precludes startReconnect() from interfering with the fast path
@@ -327,14 +464,21 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
                     if (t instanceof TransactionCommitSuccess) {
                         ret.set(Boolean.TRUE);
                     } else if (t instanceof RequestFailure) {
-                        ret.setException(((RequestFailure<?, ?>) t).getCause());
+                        final Throwable cause = ((RequestFailure<?, ?>) t).getCause().unwrap();
+                        if (cause instanceof ClosedTransactionException) {
+                            // This is okay, as it indicates the transaction has been completed. It can happen
+                            // when we lose connectivity with the backend after it has received the request.
+                            ret.set(Boolean.TRUE);
+                        } else {
+                            ret.setException(cause);
+                        }
                     } else {
                         ret.setException(new IllegalStateException("Unhandled response " + t.getClass()));
                     }
 
                     // This is a terminal request, hence we do not need to record it
                     LOG.debug("Transaction {} directCommit completed", this);
-                    parent.completeTransaction(this);
+                    enqueuePurge();
                 });
 
                 return ret;
@@ -346,6 +490,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     }
 
     final void canCommit(final VotingFuture<?> ret) {
+        checkReadWrite();
         checkSealed();
 
         // Precludes startReconnect() from interfering with the fast path
@@ -357,7 +502,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
                     if (t instanceof TransactionCanCommitSuccess) {
                         ret.voteYes();
                     } else if (t instanceof RequestFailure) {
-                        ret.voteNo(((RequestFailure<?, ?>) t).getCause());
+                        ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
                     } else {
                         ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
                     }
@@ -379,6 +524,7 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     }
 
     final void preCommit(final VotingFuture<?> ret) {
+        checkReadWrite();
         checkSealed();
 
         final TransactionRequest<?> req = new TransactionPreCommitRequest(getIdentifier(), nextSequence(),
@@ -387,33 +533,87 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
             if (t instanceof TransactionPreCommitSuccess) {
                 ret.voteYes();
             } else if (t instanceof RequestFailure) {
-                ret.voteNo(((RequestFailure<?, ?>) t).getCause());
+                ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
             } else {
                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
             }
 
-            recordSuccessfulRequest(req);
-            LOG.debug("Transaction {} preCommit completed", this);
+            onPreCommitComplete(req);
         });
     }
 
+    private void onPreCommitComplete(final TransactionRequest<?> req) {
+        /*
+         * The backend has agreed that the transaction has entered PRE_COMMIT phase, meaning it will be committed
+         * to storage after the timeout completes.
+         *
+         * All state has been replicated to the backend, hence we do not need to keep it around. Retain only
+         * the precommit request, so we know which request to use for resync.
+         */
+        LOG.debug("Transaction {} preCommit completed, clearing successfulRequests", this);
+        successfulRequests.clear();
+
+        // TODO: this works, but can contain some useless state (like batched operations). Create an empty
+        //       equivalent of this request and store that.
+        recordSuccessfulRequest(req);
+    }
+
     final void doCommit(final VotingFuture<?> ret) {
+        checkReadWrite();
         checkSealed();
 
         sendRequest(new TransactionDoCommitRequest(getIdentifier(), nextSequence(), localActor()), t -> {
             if (t instanceof TransactionCommitSuccess) {
                 ret.voteYes();
             } else if (t instanceof RequestFailure) {
-                ret.voteNo(((RequestFailure<?, ?>) t).getCause());
+                ret.voteNo(((RequestFailure<?, ?>) t).getCause().unwrap());
             } else {
                 ret.voteNo(new IllegalStateException("Unhandled response " + t.getClass()));
             }
 
             LOG.debug("Transaction {} doCommit completed", this);
+
+            // Needed for ProxyHistory$Local data tree rebase points.
             parent.completeTransaction(this);
+
+            enqueuePurge();
         });
     }
 
+    private void enqueuePurge() {
+        enqueuePurge(null);
+    }
+
+    final void enqueuePurge(final Consumer<Response<?, ?>> callback) {
+        // Purge request are dispatched internally, hence should not wait
+        enqueuePurge(callback, parent.currentTime());
+    }
+
+    final void enqueuePurge(final Consumer<Response<?, ?>> callback, final long enqueuedTicks) {
+        LOG.debug("{}: initiating purge", this);
+
+        final State prev = state;
+        if (prev instanceof SuccessorState) {
+            ((SuccessorState) prev).setDone();
+        } else {
+            final boolean success = STATE_UPDATER.compareAndSet(this, prev, DONE);
+            if (!success) {
+                LOG.warn("{}: moved from state {} while we were purging it", this, prev);
+            }
+        }
+
+        successfulRequests.clear();
+
+        enqueueRequest(new TransactionPurgeRequest(getIdentifier(), nextSequence(), localActor()), resp -> {
+            LOG.debug("{}: purge completed", this);
+            parent.purgeTransaction(this);
+
+            if (callback != null) {
+                callback.accept(resp);
+            }
+        }, enqueuedTicks);
+    }
+
     // Called with the connection unlocked
     final synchronized void startReconnect() {
         // At this point canCommit/directCommit are blocked, we assert a new successor state, retrieving the previous
@@ -432,23 +632,40 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
     }
 
     // Called with the connection locked
-    final void replayMessages(final AbstractProxyTransaction successor,
-            final Iterable<ConnectionEntry> enqueuedEntries) {
+    final void replayMessages(final ProxyHistory successorHistory, final Iterable<ConnectionEntry> enqueuedEntries) {
         final SuccessorState local = getSuccessorState();
+        final State prevState = local.getPrevState();
+
+        final AbstractProxyTransaction successor = successorHistory.createTransactionProxy(getIdentifier(),
+            isSnapshotOnly(), local.isDone());
+        LOG.debug("{} created successor {}", this, successor);
         local.setSuccessor(successor);
 
         // Replay successful requests first
-        for (Object obj : successfulRequests) {
-            if (obj instanceof TransactionRequest) {
-                LOG.debug("Forwarding successful request {} to successor {}", obj, successor);
-                successor.handleForwardedRemoteRequest((TransactionRequest<?>) obj, null);
-            } else {
-                Verify.verify(obj instanceof IncrementSequence);
-                successor.incrementSequence(((IncrementSequence) obj).getDelta());
+        if (!successfulRequests.isEmpty()) {
+            // We need to find a good timestamp to use for successful requests, as we do not want to time them out
+            // nor create timing inconsistencies in the queue -- requests are expected to be ordered by their enqueue
+            // time. We will pick the time of the first entry available. If there is none, we will just use current
+            // time, as all other requests will get enqueued afterwards.
+            final ConnectionEntry firstInQueue = Iterables.getFirst(enqueuedEntries, null);
+            final long now = firstInQueue != null ? firstInQueue.getEnqueuedTicks() : parent.currentTime();
+
+            for (Object obj : successfulRequests) {
+                if (obj instanceof TransactionRequest) {
+                    LOG.debug("Forwarding successful request {} to successor {}", obj, successor);
+                    successor.doReplayRequest((TransactionRequest<?>) obj, resp -> { }, now);
+                } else {
+                    Verify.verify(obj instanceof IncrementSequence);
+                    final IncrementSequence increment = (IncrementSequence) obj;
+                    successor.doReplayRequest(new IncrementTransactionSequenceRequest(getIdentifier(),
+                        increment.getSequence(), localActor(), isSnapshotOnly(), increment.getDelta()), resp -> { },
+                        now);
+                    LOG.debug("Incrementing sequence {} to successor {}", obj, successor);
+                }
             }
+            LOG.debug("{} replayed {} successful requests", getIdentifier(), successfulRequests.size());
+            successfulRequests.clear();
         }
-        LOG.debug("{} replayed {} successful requests", getIdentifier(), successfulRequests.size());
-        successfulRequests.clear();
 
         // Now replay whatever is in the connection
         final Iterator<ConnectionEntry> it = enqueuedEntries.iterator();
@@ -458,8 +675,8 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
 
             if (getIdentifier().equals(req.getTarget())) {
                 Verify.verify(req instanceof TransactionRequest, "Unhandled request %s", req);
-                LOG.debug("Forwarding queued request{} to successor {}", req, successor);
-                successor.handleForwardedRemoteRequest((TransactionRequest<?>) req, e.getCallback());
+                LOG.debug("Replaying queued request {} to successor {}", req, successor);
+                successor.doReplayRequest((TransactionRequest<?>) req, e.getCallback(), e.getEnqueuedTicks());
                 it.remove();
             }
         }
@@ -469,11 +686,30 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
          * reconnecting have been forced to slow paths, which will be unlocked once we unblock the state latch
          * at the end of this method.
          */
-        final State prevState = local.getPrevState();
         if (SEALED.equals(prevState)) {
             LOG.debug("Proxy {} reconnected while being sealed, propagating state to successor {}", this, successor);
             flushState(successor);
-            successor.seal();
+            successor.ensureSealed();
+        }
+    }
+
+    /**
+     * Invoked from {@link #replayMessages(AbstractProxyTransaction, Iterable)} to have successor adopt an in-flight
+     * request.
+     *
+     * <p>
+     * Note: this method is invoked by the predecessor on the successor.
+     *
+     * @param request Request which needs to be forwarded
+     * @param callback Callback to be invoked once the request completes
+     * @param enqueuedTicks ticker-based time stamp when the request was enqueued
+     */
+    private void doReplayRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback,
+            final long enqueuedTicks) {
+        if (request instanceof AbstractLocalTransactionRequest) {
+            handleReplayedLocalRequest((AbstractLocalTransactionRequest<?>) request, callback, enqueuedTicks);
+        } else {
+            handleReplayedRemoteRequest(request, callback, enqueuedTicks);
         }
     }
 
@@ -493,9 +729,12 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
      * @param request Request to be forwarded
      * @param callback Original callback
      */
-    final void replayRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
-        final AbstractProxyTransaction successor = getSuccessorState().getSuccessor();
+    final void forwardRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback) {
+        forwardToSuccessor(getSuccessorState().getSuccessor(), request, callback);
+    }
 
+    final void forwardToSuccessor(final AbstractProxyTransaction successor, final TransactionRequest<?> request,
+            final Consumer<Response<?, ?>> callback) {
         if (successor instanceof LocalProxyTransaction) {
             forwardToLocal((LocalProxyTransaction)successor, request, callback);
         } else if (successor instanceof RemoteProxyTransaction) {
@@ -505,49 +744,72 @@ abstract class AbstractProxyTransaction implements Identifiable<TransactionIdent
         }
     }
 
-    abstract void doDelete(final YangInstanceIdentifier path);
+    final void replayRequest(final TransactionRequest<?> request, final Consumer<Response<?, ?>> callback,
+            final long enqueuedTicks) {
+        getSuccessorState().getSuccessor().doReplayRequest(request, callback, enqueuedTicks);
+    }
 
-    abstract void doMerge(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data);
+    abstract boolean isSnapshotOnly();
 
-    abstract void doWrite(final YangInstanceIdentifier path, final NormalizedNode<?, ?> data);
+    abstract void doDelete(YangInstanceIdentifier path);
 
-    abstract CheckedFuture<Boolean, ReadFailedException> doExists(final YangInstanceIdentifier path);
+    abstract void doMerge(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
 
-    abstract CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(
-            final YangInstanceIdentifier path);
+    abstract void doWrite(YangInstanceIdentifier path, NormalizedNode<?, ?> data);
 
-    abstract void doSeal();
+    abstract CheckedFuture<Boolean, ReadFailedException> doExists(YangInstanceIdentifier path);
 
-    abstract void doAbort();
+    abstract CheckedFuture<Optional<NormalizedNode<?, ?>>, ReadFailedException> doRead(YangInstanceIdentifier path);
+
+    abstract void doSeal();
 
     @GuardedBy("this")
     abstract void flushState(AbstractProxyTransaction successor);
 
+    abstract TransactionRequest<?> abortRequest();
+
     abstract TransactionRequest<?> commitRequest(boolean coordinated);
 
     /**
-     * Invoked from {@link RemoteProxyTransaction} when it replays its successful requests to its successor. There is
-     * no equivalent of this call from {@link LocalProxyTransaction} because it does not send a request until all
-     * operations are packaged in the message.
+     * Replay a request originating in this proxy to a successor remote proxy.
+     */
+    abstract void forwardToRemote(RemoteProxyTransaction successor, TransactionRequest<?> request,
+            Consumer<Response<?, ?>> callback);
+
+    /**
+     * Replay a request originating in this proxy to a successor local proxy.
+     */
+    abstract void forwardToLocal(LocalProxyTransaction successor, TransactionRequest<?> request,
+            Consumer<Response<?, ?>> callback);
+
+    /**
+     * Invoked from {@link LocalProxyTransaction} when it replays its successful requests to its successor.
      *
      * <p>
      * Note: this method is invoked by the predecessor on the successor.
      *
      * @param request Request which needs to be forwarded
      * @param callback Callback to be invoked once the request completes
+     * @param enqueuedTicks Time stamp to use for enqueue time
      */
-    abstract void handleForwardedRemoteRequest(TransactionRequest<?> request,
-            @Nullable Consumer<Response<?, ?>> callback);
+    abstract void handleReplayedLocalRequest(AbstractLocalTransactionRequest<?> request,
+            @Nullable Consumer<Response<?, ?>> callback, long enqueuedTicks);
 
     /**
-     * Replay a request originating in this proxy to a successor remote proxy.
+     * Invoked from {@link RemoteProxyTransaction} when it replays its successful requests to its successor.
+     *
+     * <p>
+     * Note: this method is invoked by the predecessor on the successor.
+     *
+     * @param request Request which needs to be forwarded
+     * @param callback Callback to be invoked once the request completes
+     * @param enqueuedTicks Time stamp to use for enqueue time
      */
-    abstract void forwardToRemote(RemoteProxyTransaction successor, TransactionRequest<?> request,
-            Consumer<Response<?, ?>> callback);
+    abstract void handleReplayedRemoteRequest(TransactionRequest<?> request,
+            @Nullable Consumer<Response<?, ?>> callback, long enqueuedTicks);
 
-    /**
-     * Replay a request originating in this proxy to a successor local proxy.
-     */
-    abstract void forwardToLocal(LocalProxyTransaction successor, TransactionRequest<?> request,
-            Consumer<Response<?, ?>> callback);
+    @Override
+    public final String toString() {
+        return MoreObjects.toStringHelper(this).add("identifier", getIdentifier()).add("state", state).toString();
+    }
 }