cassandra/test/unit/org/apache/cassandra/repair/FailingRepairFuzzTest.java

/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

package org.apache.cassandra.repair;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

import com.google.common.collect.ImmutableList;
import org.junit.Test;

import accord.utilsfork.Gen;
import accord.utilsfork.Gens;
import org.apache.cassandra.config.DatabaseDescriptor;
import org.apache.cassandra.config.RetrySpec;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.repair.messages.RepairOption;
import org.apache.cassandra.repair.state.Completable;
import org.apache.cassandra.streaming.StreamEventHandler;
import org.apache.cassandra.streaming.StreamState;
import org.apache.cassandra.utils.Closeable;
import org.assertj.core.api.AbstractStringAssert;
import org.assertj.core.api.Assertions;

import static accord.utilsfork.Property.qt;

public class FailingRepairFuzzTest extends FuzzTestBase
{
    private enum RepairJobStage { VALIDATION, SYNC }

    @Test
    public void failingRepair()
    {
        // to avoid unlucky timing issues, retry until success; given enough retries we should eventually become success
        DatabaseDescriptor.getRepairRetrySpec().maxAttempts = new RetrySpec.MaxAttempt(Integer.MAX_VALUE);
        Gen<RepairJobStage> stageGen = Gens.enums().all(RepairJobStage.class);
        qt().withPure(false).withExamples(10).check(rs -> {
            Cluster cluster = new Cluster(rs);
            enableMessageFaults(cluster);

            Gen<Cluster.Node> coordinatorGen = Gens.pick(cluster.nodes.keySet()).map(cluster.nodes::get);

            List<Closeable> closeables = new ArrayList<>();
            for (int example = 0; example < 100; example++)
            {
                Cluster.Node coordinator = coordinatorGen.next(rs);

                // exclude accord repair as this test breaks validation/sync; which accord doesn't have
                RepairOption options = repairOption(rs, coordinator, KEYSPACE, TABLES);
                RepairCoordinator repair = coordinator.repair(KEYSPACE, options, false);
                repair.run();
                InetAddressAndPort failingAddress = pickParticipant(rs, coordinator, repair);
                Cluster.Node failingNode = cluster.nodes.get(failingAddress);
                RepairJobStage stage = stageGen.next(rs);
                // because of local syncs reaching out to the failing address, a different address may actually be what failed
                Set<InetAddressAndPort> syncFailedAddresses = new HashSet<>();
                switch (stage)
                {
                    case VALIDATION:
                    {
                        closeables.add(failingNode.doValidation((cfs, validator) -> {
                            long delayNanos = rs.nextLong(TimeUnit.MILLISECONDS.toNanos(5), TimeUnit.MINUTES.toNanos(1));
                            cluster.unorderedScheduled.schedule(() -> validator.fail(new SimulatedFault("Validation failed")), delayNanos, TimeUnit.NANOSECONDS);
                        }));
                    }
                    break;
                    case SYNC:
                    {
                        closeables.add(failingNode.doValidation((cfs, validator) -> addMismatch(rs, cfs, validator)));
                        List<InetAddressAndPort> addresses = ImmutableList.<InetAddressAndPort>builder().add(coordinator.addressAndPort).addAll(repair.state.getNeighborsAndRanges().participants).build();
                        for (InetAddressAndPort address : addresses)
                        {
                            closeables.add(cluster.nodes.get(address).doSync(plan -> {
                                long delayNanos = rs.nextLong(TimeUnit.SECONDS.toNanos(5), TimeUnit.MINUTES.toNanos(10));
                                cluster.unorderedScheduled.schedule(() -> {
                                    if (address == failingAddress || plan.getCoordinator().getPeers().contains(failingAddress))
                                    {
                                        syncFailedAddresses.add(address);
                                        SimulatedFault fault = new SimulatedFault("Sync failed");
                                        for (StreamEventHandler handler : plan.handlers())
                                            handler.onFailure(fault);
                                    }
                                    else
                                    {
                                        StreamState success = new StreamState(plan.planId(), plan.streamOperation(), Collections.emptySet());
                                        for (StreamEventHandler handler : plan.handlers())
                                            handler.onSuccess(success);
                                    }
                                }, delayNanos, TimeUnit.NANOSECONDS);
                                return null;
                            }));
                        }
                    }
                    break;
                    default:
                        throw new IllegalArgumentException("Unknown stage: " + stage);
                }

                cluster.processAll();
                Assertions.assertThat(repair.state.isComplete()).describedAs("Repair job did not complete, and no work is pending...").isTrue();
                Assertions.assertThat(repair.state.getResult().kind).describedAs("Unexpected state: %s -> %s; example %d", repair.state, repair.state.getResult(), example).isEqualTo(Completable.Result.Kind.FAILURE);
                switch (stage)
                {
                    case VALIDATION:
                    {
                        // Got VALIDATION_REQ failure from Ero2.MJ.N8kkw2.w3iFYdDw.HJiVYC32.mWb.b.xwi3tZ.s5k1l.mb.asTy_7QmQ.Q3.u.kjgh.GKjx.g1aKfkjB.YlyKg9.DyQszn7F.Ox2DMYIph.xlgH.EV.A9yEz2J.l6UHdC.C6FYLXE.J0CNHBH./[4905:e9f:2f00:e418:baac:b8d9:9ff9:6604]:33363: UNKNOWN"
                        Assertions.assertThat(repair.state.getResult().message)
                                  .describedAs("Unexpected state: %s -> %s; example %d", repair.state, repair.state.getResult(), example)
                                  // ValidationResponse with null tree seen
                                  .containsAnyOf("Validation failed in " + failingAddress,
                                                 // ack was dropped and on retry the participate detected dup so rejected as the task failed
                                                 "Got VALIDATION_REQ failure from " + failingAddress + ": UNKNOWN");
                    }
                    break;
                    case SYNC:
                        AbstractStringAssert<?> a = Assertions.assertThat(repair.state.getResult().message).describedAs("Unexpected state: %s -> %s; example %d", repair.state, repair.state.getResult(), example);
                        // SymmetricRemoteSyncTask + AsymmetricRemoteSyncTask
                        // ... Sync failed between /[81fc:714:2c56:a2d3:faf3:eb7c:e4dd:cb9e]:54401 and /220.3.10.72:21402
                        // LocalSyncTask
                        // ... failed with error Sync failed
                        // Dedup nack, but may be remote or local sync!
                        // ... Got SYNC_REQ failure from ...: UNKNOWN
                        String failingMsg = repair.state.getResult().message;
                        if (failingMsg.contains("Sync failed between"))
                        {
                            a.contains("Sync failed between").contains(failingAddress.toString());
                        }
                        else if (failingMsg.contains("Got SYNC_REQ failure from"))
                        {
                            Assertions.assertThat(syncFailedAddresses).isNotEmpty();
                            a.containsAnyOf(syncFailedAddresses.stream().map(s -> "Got SYNC_REQ failure from " + s + ": UNKNOWN").collect(Collectors.toList()).toArray(String[]::new));
                        }
                        else
                        {
                            a.contains("failed with error Sync failed");
                        }
                        break;
                    default:
                        throw new IllegalArgumentException("Unknown stage: " + stage);
                }
                assertParticipateResult(cluster, repair, Completable.Result.Kind.FAILURE);
                closeables.forEach(Closeable::close);
                closeables.clear();
            }
        });
    }
}