Remove last accepted slot (elastic#10)

Relates to elastic/elasticsearch-formal-models#21

Author: ywelsch

core/src/main/java/org/elasticsearch/discovery/zen2/ConsensusState.java

@@ -25,7 +25,6 @@
 import org.elasticsearch.discovery.zen2.Messages.ApplyCommit;
 import org.elasticsearch.discovery.zen2.Messages.PublishRequest;
 import org.elasticsearch.discovery.zen2.Messages.PublishResponse;
-import org.elasticsearch.discovery.zen2.Messages.SlotTermDiff;
 import org.elasticsearch.discovery.zen2.Messages.Vote;
 
 import java.util.HashMap;
@@ -48,15 +47,15 @@ public class ConsensusState<T extends ConsensusState.CommittedState> extends Abs
     // persisted state
     private long currentTerm;
     private T committedState;
-    private Optional<SlotTermDiff<T>> acceptedState;
+    private Optional<AcceptedState<T>> acceptedState;
     // transient state
     private boolean electionWon;
     private boolean electionValueForced;
     private NodeCollection joinVotes;
     private boolean publishPermitted;
     private NodeCollection publishVotes;
 
-    public ConsensusState(Settings settings, long currentTerm, T committedState, Optional<SlotTermDiff<T>> acceptedState,
+    public ConsensusState(Settings settings, long currentTerm, T committedState, Optional<AcceptedState<T>> acceptedState,
                           Persistence<T> persistence) {
         // TODO idea: just pass in a Persistence and let it provide the persisted state.
 
@@ -76,8 +75,7 @@ public ConsensusState(Settings settings, long currentTerm, T committedState, Opt
         this.publishVotes = new NodeCollection();
 
         assert currentTerm >= 0;
-        assert acceptedState.isPresent() == false || acceptedState.get().getTerm() <= currentTerm;
-        assert acceptedState.isPresent() == false || acceptedState.get().getSlot() <= firstUncommittedSlot();
+        assert lastAcceptedTerm() <= currentTerm;
     }
 
     public T getCommittedState() {
@@ -98,8 +96,8 @@ public long firstUncommittedSlot() {
         return committedState.getSlot() + 1;
     }
 
-    public long lastAcceptedTermInSlot() {
-        if (acceptedState.isPresent() && firstUncommittedSlot() == acceptedState.get().getSlot()) {
+    public long lastAcceptedTerm() {
+        if (acceptedState.isPresent()) {
             return acceptedState.get().getTerm();
         } else {
             return NO_TERM;
@@ -131,7 +129,7 @@ public Vote handleStartVote(long newTerm) {
         publishPermitted = true;
         publishVotes = new NodeCollection();
 
-        return new Vote(firstUncommittedSlot(), currentTerm, lastAcceptedTermInSlot());
+        return new Vote(firstUncommittedSlot(), currentTerm, lastAcceptedTerm());
     }
 
     /**
@@ -155,7 +153,7 @@ public Optional<PublishRequest<T>> handleVote(DiscoveryNode sourceNode, Vote vot
                 firstUncommittedSlot());
         }
         if (vote.getFirstUncommittedSlot() == firstUncommittedSlot() && vote.getLastAcceptedTerm() != NO_TERM) {
-            final long lastAcceptedTermInSlot = lastAcceptedTermInSlot();
+            final long lastAcceptedTermInSlot = lastAcceptedTerm();
             if (vote.getLastAcceptedTerm() > lastAcceptedTermInSlot) {
                 logger.debug("handleVote: ignored vote as voter has better last accepted term (expected: <=[{}], actual: [{}])",
                     lastAcceptedTermInSlot, vote.getLastAcceptedTerm());
@@ -212,8 +210,9 @@ public PublishResponse handlePublishRequest(PublishRequest<T> publishRequest) {
 
         logger.trace("handlePublishRequest: storing publish request for slot [{}] and term [{}]",
             publishRequest.getSlot(), publishRequest.getTerm());
-        persistence.persistAcceptedState(publishRequest);
-        acceptedState = Optional.of(publishRequest);
+        final AcceptedState<T> termDiff = publishRequest.getAcceptedState();
+        persistence.persistAcceptedState(termDiff);
+        acceptedState = Optional.of(termDiff);
 
         return new PublishResponse(publishRequest.getSlot(), publishRequest.getTerm());
     }
@@ -264,11 +263,11 @@ public Optional<ApplyCommit> handlePublishResponse(DiscoveryNode sourceNode, Pub
      * @throws IllegalArgumentException if the arguments were incompatible with the current state of this object.
      */
     public void handleCommit(ApplyCommit applyCommit) {
-        if (applyCommit.getTerm() != lastAcceptedTermInSlot()) {
+        if (applyCommit.getTerm() != lastAcceptedTerm()) {
             logger.debug("handleCommit: ignored commit request due to term mismatch (expected: [{}], actual: [{}])",
-                lastAcceptedTermInSlot(), applyCommit.getTerm());
+                lastAcceptedTerm(), applyCommit.getTerm());
             throw new IllegalArgumentException("incoming term " + applyCommit.getTerm() + " does not match last accepted term " +
-                lastAcceptedTermInSlot());
+                lastAcceptedTerm());
         }
         if (applyCommit.getSlot() != firstUncommittedSlot()) {
             logger.debug("handleCommit: ignored commit request due to slot mismatch (expected: [{}], actual: [{}])",
@@ -280,13 +279,13 @@ public void handleCommit(ApplyCommit applyCommit) {
         logger.trace("handleCommit: applying commit request for slot [{}]", applyCommit.getSlot());
 
         assert acceptedState.isPresent();
-        assert acceptedState.get().getSlot() == committedState.getSlot() + 1;
         final T newCommittedState = acceptedState.get().getDiff().apply(committedState);
         logger.trace("handleCommit: newCommittedState = [{}]", newCommittedState);
         assert newCommittedState.getSlot() == committedState.getSlot() + 1;
 
         persistence.persistCommittedState(newCommittedState);
         committedState = newCommittedState;
+        acceptedState = Optional.empty();
         publishPermitted = true;
         electionValueForced = false;
         publishVotes = new NodeCollection();
@@ -318,6 +317,7 @@ public void applyCatchup(T newCommittedState) {
         logger.debug("applyCatchup: applying catch up for slot [{}]", newCommittedState.getSlot());
         persistence.persistCommittedState(newCommittedState);
         committedState = newCommittedState;
+        acceptedState = Optional.empty();
         electionValueForced = false;
         joinVotes = new NodeCollection();
         electionWon = false;
@@ -360,12 +360,57 @@ public interface CommittedState {
         NodeCollection getVotingNodes();
     }
 
+    public static class AcceptedState<T> {
+        protected final long term;
+        protected final Diff<T> diff;
+
+        public AcceptedState(long term, Diff<T> diff) {
+            this.term = term;
+            this.diff = diff;
+        }
+
+        public long getTerm() {
+            return term;
+        }
+
+        public Diff<T> getDiff() {
+            return diff;
+        }
+
+        @Override
+        public boolean equals(Object o) {
+            if (this == o) return true;
+            if (o == null || getClass() != o.getClass()) return false;
+
+            AcceptedState<?> termDiff = (AcceptedState<?>) o;
+
+            if (term != termDiff.term) return false;
+            return diff.equals(termDiff.diff);
+        }
+
+        @Override
+        public int hashCode() {
+            int result = (int) (term ^ (term >>> 32));
+            result = 31 * result + diff.hashCode();
+            return result;
+        }
+
+
+        @Override
+        public String toString() {
+            return "AcceptedState{" +
+                "term=" + term +
+                ", diff=" + diff +
+                '}';
+        }
+    }
+
     public interface Persistence<T> {
         void persistCurrentTerm(long currentTerm);
 
         void persistCommittedState(T committedState);
 
-        void persistAcceptedState(SlotTermDiff<T> slotTermDiff);
+        void persistAcceptedState(AcceptedState<T> termDiff);
     }
 
     /**

core/src/main/java/org/elasticsearch/discovery/zen2/Messages.java

@@ -159,10 +159,11 @@ public String toString() {
         }
     }
 
-    public abstract static class SlotTermDiff<T> extends SlotTerm {
+    public static class PublishRequest<T> extends SlotTerm {
+
         protected final Diff<T> diff;
 
-        public SlotTermDiff(long slot, long term, Diff<T> diff) {
+        public PublishRequest(long slot, long term, Diff<T> diff) {
             super(slot, term);
             this.diff = diff;
         }
@@ -171,12 +172,16 @@ public Diff<T> getDiff() {
             return diff;
         }
 
+        public ConsensusState.AcceptedState<T> getAcceptedState() {
+            return new ConsensusState.AcceptedState<>(term, diff);
+        }
+
         @Override
         public boolean equals(Object o) {
             if (super.equals(o) == false) {
                 return false;
             }
-            SlotTermDiff<?> that = (SlotTermDiff<?>) o;
+            PublishRequest<?> that = (PublishRequest<?>) o;
             return diff != null ? diff.equals(that.diff) : that.diff == null;
         }
 
@@ -187,22 +192,6 @@ public int hashCode() {
             return result;
         }
 
-        @Override
-        public String toString() {
-            return "SlotTermDiff{" +
-                "slot=" + slot +
-                ", term=" + term +
-                ", diff=" + diff +
-                '}';
-        }
-    }
-
-    public static class PublishRequest<T> extends SlotTermDiff<T> {
-
-        public PublishRequest(long slot, long term, Diff<T> diff) {
-            super(slot, term, diff);
-        }
-
         @Override
         public String toString() {
             return "PublishRequest{" +

core/src/test/java/org/elasticsearch/discovery/zen2/ConsensusStateTests.java

@@ -55,7 +55,7 @@ public void persistCommittedState(ClusterState committedState) {
                 }
 
                 @Override
-                public void persistAcceptedState(Messages.SlotTermDiff<ClusterState> slotTermDiff) {
+                public void persistAcceptedState(ConsensusState.AcceptedState<ClusterState> termDiff) {
 
                 }
             });