Verify primary mode usage with assertions (#32667)

Primary terms were introduced as part of the sequence-number effort (#10708) and added in ES
5.0. Subsequent work introduced the replication tracker which lets the primary own its replication
group (#25692) to coordinate recovery and replication. The replication tracker explicitly exposes
whether it is operating in primary mode or replica mode, independent of the ShardRouting object
that's associated with a shard. During a primary relocation, for example, the primary mode is
transferred between the primary relocation source and the primary relocation target. After
transferring this so-called primary context, the old primary becomes a replication target and the
new primary the replication source, reflected in the replication tracker on both nodes. With the
most recent PR in this area (#32442), we finally have a clean transition between a shard that's
operating as a primary and issuing sequence numbers and a shard that's serving as a replication
target. The transition from one state to the other is enforced through the operation-permit system,
where we block permit acquisition during such changes and perform the transition under this
operation block, ensuring that there are no operations in progress while the transition is being
performed. This finally allows us to turn the best-effort checks that were put in place to prevent
shards from being used in the wrong way (i.e. primary as replica, or replica as primary) into hard
assertions, making it easier to catch any bugs in this area.

Author: ywelsch

server/src/main/java/org/elasticsearch/index/shard/IndexShard.java

@@ -1481,30 +1481,25 @@ private void ensureWriteAllowed(Engine.Operation.Origin origin) throws IllegalIn
             }
         } else {
             if (origin == Engine.Operation.Origin.PRIMARY) {
-                verifyPrimary();
+                assert assertPrimaryMode();
             } else {
                 assert origin == Engine.Operation.Origin.REPLICA;
-                verifyReplicationTarget();
+                assert assertReplicationTarget();
             }
             if (writeAllowedStates.contains(state) == false) {
                 throw new IllegalIndexShardStateException(shardId, state, "operation only allowed when shard state is one of " + writeAllowedStates + ", origin [" + origin + "]");
             }
         }
     }
 
-    private void verifyPrimary() {
-        if (shardRouting.primary() == false) {
-            throw new IllegalStateException("shard " + shardRouting + " is not a primary");
-        }
+    private boolean assertPrimaryMode() {
+        assert shardRouting.primary() && replicationTracker.isPrimaryMode() : "shard " + shardRouting + " is not a primary shard in primary mode";
+        return true;
     }
 
-    private void verifyReplicationTarget() {
-        final IndexShardState state = state();
-        if (shardRouting.primary() && shardRouting.active() && replicationTracker.isPrimaryMode()) {
-            // must use exception that is not ignored by replication logic. See TransportActions.isShardNotAvailableException
-            throw new IllegalStateException("active primary shard " + shardRouting + " cannot be a replication target before " +
-                "relocation hand off, state is [" + state + "]");
-        }
+    private boolean assertReplicationTarget() {
+        assert replicationTracker.isPrimaryMode() == false : "shard " + shardRouting + " in primary mode cannot be a replication target";
+        return true;
     }
 
     private void verifyNotClosed() throws IllegalIndexShardStateException {
@@ -1751,7 +1746,7 @@ public void writeIndexingBuffer() {
      * @param checkpoint   the local checkpoint for the shard
      */
     public void updateLocalCheckpointForShard(final String allocationId, final long checkpoint) {
-        verifyPrimary();
+        assert assertPrimaryMode();
         verifyNotClosed();
         replicationTracker.updateLocalCheckpoint(allocationId, checkpoint);
     }
@@ -1763,7 +1758,7 @@ public void updateLocalCheckpointForShard(final String allocationId, final long
      * @param globalCheckpoint the global checkpoint
      */
     public void updateGlobalCheckpointForShard(final String allocationId, final long globalCheckpoint) {
-        verifyPrimary();
+        assert assertPrimaryMode();
         verifyNotClosed();
         replicationTracker.updateGlobalCheckpointForShard(allocationId, globalCheckpoint);
     }
@@ -1785,7 +1780,7 @@ public void waitForOpsToComplete(final long seqNo) throws InterruptedException {
      * @param allocationId  the allocation ID of the shard for which recovery was initiated
      */
     public void initiateTracking(final String allocationId) {
-        verifyPrimary();
+        assert assertPrimaryMode();
         replicationTracker.initiateTracking(allocationId);
     }
 
@@ -1798,7 +1793,7 @@ public void initiateTracking(final String allocationId) {
      * @param localCheckpoint the current local checkpoint on the shard
      */
     public void markAllocationIdAsInSync(final String allocationId, final long localCheckpoint) throws InterruptedException {
-        verifyPrimary();
+        assert assertPrimaryMode();
         replicationTracker.markAllocationIdAsInSync(allocationId, localCheckpoint);
     }
 
@@ -1833,7 +1828,7 @@ public long getLastSyncedGlobalCheckpoint() {
      * @return a map from allocation ID to the local knowledge of the global checkpoint for that allocation ID
      */
     public ObjectLongMap<String> getInSyncGlobalCheckpoints() {
-        verifyPrimary();
+        assert assertPrimaryMode();
         verifyNotClosed();
         return replicationTracker.getInSyncGlobalCheckpoints();
     }
@@ -1843,11 +1838,12 @@ public ObjectLongMap<String> getInSyncGlobalCheckpoints() {
      * primary.
      */
     public void maybeSyncGlobalCheckpoint(final String reason) {
-        verifyPrimary();
         verifyNotClosed();
+        assert shardRouting.primary() : "only call maybeSyncGlobalCheckpoint on primary shard";
         if (replicationTracker.isPrimaryMode() == false) {
             return;
         }
+        assert assertPrimaryMode();
         // only sync if there are not operations in flight
         final SeqNoStats stats = getEngine().getSeqNoStats(replicationTracker.getGlobalCheckpoint());
         if (stats.getMaxSeqNo() == stats.getGlobalCheckpoint()) {
@@ -1873,7 +1869,7 @@ public void maybeSyncGlobalCheckpoint(final String reason) {
      * @return the replication group
      */
     public ReplicationGroup getReplicationGroup() {
-        verifyPrimary();
+        assert assertPrimaryMode();
         verifyNotClosed();
         return replicationTracker.getReplicationGroup();
     }
@@ -1885,7 +1881,7 @@ public ReplicationGroup getReplicationGroup() {
      * @param reason           the reason the global checkpoint was updated
      */
     public void updateGlobalCheckpointOnReplica(final long globalCheckpoint, final String reason) {
-        verifyReplicationTarget();
+        assert assertReplicationTarget();
         final long localCheckpoint = getLocalCheckpoint();
         if (globalCheckpoint > localCheckpoint) {
             /*
@@ -1912,8 +1908,7 @@ assert state() != IndexShardState.POST_RECOVERY && state() != IndexShardState.ST
      * @param primaryContext the sequence number context
      */
     public void activateWithPrimaryContext(final ReplicationTracker.PrimaryContext primaryContext) {
-        verifyPrimary();
-        assert shardRouting.isRelocationTarget() : "only relocation target can update allocation IDs from primary context: " + shardRouting;
+        assert shardRouting.primary() && shardRouting.isRelocationTarget() : "only primary relocation target can update allocation IDs from primary context: " + shardRouting;
         assert primaryContext.getCheckpointStates().containsKey(routingEntry().allocationId().getId()) &&
             getLocalCheckpoint() == primaryContext.getCheckpointStates().get(routingEntry().allocationId().getId()).getLocalCheckpoint();
         synchronized (mutex) {
@@ -1927,7 +1922,7 @@ public void activateWithPrimaryContext(final ReplicationTracker.PrimaryContext p
      * @return {@code true} if there is at least one shard pending in-sync, otherwise false
      */
     public boolean pendingInSync() {
-        verifyPrimary();
+        assert assertPrimaryMode();
         return replicationTracker.pendingInSync();
     }
 
@@ -2244,7 +2239,7 @@ private EngineConfig newEngineConfig() {
      */
     public void acquirePrimaryOperationPermit(ActionListener<Releasable> onPermitAcquired, String executorOnDelay, Object debugInfo) {
         verifyNotClosed();
-        verifyPrimary();
+        assert shardRouting.primary() : "acquirePrimaryOperationPermit should only be called on primary shard: " + shardRouting;
 
         indexShardOperationPermits.acquire(onPermitAcquired, executorOnDelay, false, debugInfo);
     }
@@ -2294,7 +2289,6 @@ public void acquireReplicaOperationPermit(final long opPrimaryTerm, final long g
                                               final ActionListener<Releasable> onPermitAcquired, final String executorOnDelay,
                                               final Object debugInfo) {
         verifyNotClosed();
-        verifyReplicationTarget();
         if (opPrimaryTerm > pendingPrimaryTerm) {
             synchronized (mutex) {
                 if (opPrimaryTerm > pendingPrimaryTerm) {
@@ -2347,6 +2341,7 @@ public void onResponse(final Releasable releasable) {
                                     operationPrimaryTerm);
                             onPermitAcquired.onFailure(new IllegalStateException(message));
                         } else {
+                            assert assertReplicationTarget();
                             try {
                                 updateGlobalCheckpointOnReplica(globalCheckpoint, "operation");
                             } catch (Exception e) {

server/src/test/java/org/elasticsearch/index/shard/IndexShardTests.java

@@ -31,6 +31,7 @@
 import org.apache.lucene.store.FilterDirectory;
 import org.apache.lucene.store.IOContext;
 import org.apache.lucene.util.Constants;
+import org.elasticsearch.Assertions;
 import org.elasticsearch.Version;
 import org.elasticsearch.action.ActionListener;
 import org.elasticsearch.action.admin.indices.flush.FlushRequest;
@@ -559,28 +560,20 @@ public void testOperationPermitsOnPrimaryShards() throws InterruptedException, E
             ShardRouting primaryRouting = newShardRouting(replicaRouting.shardId(), replicaRouting.currentNodeId(), null,
                 true, ShardRoutingState.STARTED, replicaRouting.allocationId());
             final long newPrimaryTerm = indexShard.getPendingPrimaryTerm() + between(1, 1000);
+            CountDownLatch latch = new CountDownLatch(1);
             indexShard.updateShardState(primaryRouting, newPrimaryTerm, (shard, listener) -> {
                     assertThat(TestTranslog.getCurrentTerm(getTranslog(indexShard)), equalTo(newPrimaryTerm));
+                    latch.countDown();
                 }, 0L,
                 Collections.singleton(indexShard.routingEntry().allocationId().getId()),
                 new IndexShardRoutingTable.Builder(indexShard.shardId()).addShard(primaryRouting).build(),
                 Collections.emptySet());
+            latch.await();
         } else {
             indexShard = newStartedShard(true);
         }
         final long primaryTerm = indexShard.getPendingPrimaryTerm();
         assertEquals(0, indexShard.getActiveOperationsCount());
-        if (indexShard.routingEntry().isRelocationTarget() == false) {
-            try {
-                final PlainActionFuture<Releasable> permitAcquiredFuture = new PlainActionFuture<>();
-                indexShard.acquireReplicaOperationPermit(primaryTerm, indexShard.getGlobalCheckpoint(), permitAcquiredFuture,
-                    ThreadPool.Names.WRITE, "");
-                permitAcquiredFuture.actionGet();
-                fail("shard shouldn't accept operations as replica");
-            } catch (IllegalStateException ignored) {
-
-            }
-        }
         Releasable operation1 = acquirePrimaryOperationPermitBlockingly(indexShard);
         assertEquals(1, indexShard.getActiveOperationsCount());
         Releasable operation2 = acquirePrimaryOperationPermitBlockingly(indexShard);
@@ -589,6 +582,22 @@ public void testOperationPermitsOnPrimaryShards() throws InterruptedException, E
         Releasables.close(operation1, operation2);
         assertEquals(0, indexShard.getActiveOperationsCount());
 
+        if (Assertions.ENABLED && indexShard.routingEntry().isRelocationTarget() == false) {
+            assertThat(expectThrows(AssertionError.class, () -> indexShard.acquireReplicaOperationPermit(primaryTerm,
+                indexShard.getGlobalCheckpoint(), new ActionListener<Releasable>() {
+                    @Override
+                    public void onResponse(Releasable releasable) {
+                        fail();
+                    }
+
+                    @Override
+                    public void onFailure(Exception e) {
+                        fail();
+                    }
+                },
+                ThreadPool.Names.WRITE, "")).getMessage(), containsString("in primary mode cannot be a replication target"));
+        }
+
         closeShards(indexShard);
     }
 
@@ -646,11 +655,11 @@ public void testOperationPermitOnReplicaShards() throws Exception {
         logger.info("shard routing to {}", shardRouting);
 
         assertEquals(0, indexShard.getActiveOperationsCount());
-        if (shardRouting.primary() == false) {
-            final IllegalStateException e =
-                    expectThrows(IllegalStateException.class,
-                        () -> indexShard.acquirePrimaryOperationPermit(null, ThreadPool.Names.INDEX, ""));
-            assertThat(e, hasToString(containsString("shard " + shardRouting + " is not a primary")));
+        if (shardRouting.primary() == false && Assertions.ENABLED) {
+            final AssertionError e =
+                    expectThrows(AssertionError.class,
+                        () -> indexShard.acquirePrimaryOperationPermit(null, ThreadPool.Names.WRITE, ""));
+            assertThat(e, hasToString(containsString("acquirePrimaryOperationPermit should only be called on primary shard")));
         }
 
         final long primaryTerm = indexShard.getPendingPrimaryTerm();