/*
 * 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.simulator.test;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.EnumMap;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Supplier;

import org.junit.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import accord.utils.Invariants;

import org.apache.cassandra.dht.Murmur3Partitioner;
import org.apache.cassandra.distributed.Cluster;
import org.apache.cassandra.distributed.api.ConsistencyLevel;
import org.apache.cassandra.distributed.api.IInstanceConfig;
import org.apache.cassandra.distributed.api.IInvokableInstance;
import org.apache.cassandra.distributed.api.IIsolatedExecutor;
import org.apache.cassandra.distributed.api.SimpleQueryResult;
import org.apache.cassandra.distributed.impl.Query;
import org.apache.cassandra.distributed.shared.WithProperties;
import org.apache.cassandra.harry.SchemaSpec;
import org.apache.cassandra.harry.execution.CompiledStatement;
import org.apache.cassandra.harry.execution.DataTracker;
import org.apache.cassandra.harry.execution.QueryBuildingVisitExecutor;
import org.apache.cassandra.harry.gen.EntropySource;
import org.apache.cassandra.harry.gen.Generator;
import org.apache.cassandra.harry.gen.OperationsGenerators;
import org.apache.cassandra.harry.gen.SchemaGenerators;
import org.apache.cassandra.harry.gen.rng.JdkRandomEntropySource;
import org.apache.cassandra.harry.model.Model;
import org.apache.cassandra.harry.model.QuiescentChecker;
import org.apache.cassandra.harry.model.TokenPlacementModel;
import org.apache.cassandra.harry.op.Operations;
import org.apache.cassandra.harry.op.Visit;
import org.apache.cassandra.locator.InetAddressAndPort;
import org.apache.cassandra.net.Verb;
import org.apache.cassandra.schema.ReplicationParams;
import org.apache.cassandra.simulator.Action;
import org.apache.cassandra.simulator.ActionList;
import org.apache.cassandra.simulator.ActionSchedule;
import org.apache.cassandra.simulator.Actions;
import org.apache.cassandra.simulator.AlwaysDeliverNetworkScheduler;
import org.apache.cassandra.simulator.ClusterSimulation;
import org.apache.cassandra.simulator.Debug;
import org.apache.cassandra.simulator.FixedLossNetworkScheduler;
import org.apache.cassandra.simulator.FutureActionScheduler;
import org.apache.cassandra.simulator.OrderOn;
import org.apache.cassandra.simulator.RandomSource;
import org.apache.cassandra.simulator.RunnableActionScheduler;
import org.apache.cassandra.simulator.Simulation;
import org.apache.cassandra.simulator.SimulationException;
import org.apache.cassandra.simulator.SimulationRunner;
import org.apache.cassandra.simulator.SimulatorUtils;
import org.apache.cassandra.simulator.cluster.ClusterActionListener.NoOpListener;
import org.apache.cassandra.simulator.cluster.ClusterActions;
import org.apache.cassandra.simulator.cluster.ClusterActions.Options;
import org.apache.cassandra.simulator.systems.Failures;
import org.apache.cassandra.simulator.systems.InterceptedExecution;
import org.apache.cassandra.simulator.systems.InterceptingExecutor;
import org.apache.cassandra.simulator.systems.SimulatedActionCallable;
import org.apache.cassandra.simulator.systems.SimulatedActionTask;
import org.apache.cassandra.simulator.systems.SimulatedSystems;
import org.apache.cassandra.simulator.systems.SimulatedTime;
import org.apache.cassandra.simulator.utils.KindOfSequence;
import org.apache.cassandra.tcm.ClusterMetadata;
import org.apache.cassandra.tcm.ClusterMetadataService;
import org.apache.cassandra.tcm.Startup;
import org.apache.cassandra.tcm.membership.NodeState;
import org.apache.cassandra.tcm.sequences.SingleNodeSequences;
import org.apache.cassandra.tcm.transformations.PrepareJoin;
import org.apache.cassandra.utils.CloseableIterator;
import org.apache.cassandra.utils.concurrent.UncheckedInterruptedException;

import picocli.CommandLine;
import picocli.CommandLine.Command;
import picocli.CommandLine.Option;

import static java.util.concurrent.TimeUnit.SECONDS;
import static org.apache.cassandra.distributed.api.ConsistencyLevel.ALL;
import static org.apache.cassandra.harry.model.TokenPlacementModel.constantLookup;
import static org.apache.cassandra.simulator.ActionSchedule.Mode.UNLIMITED;
import static org.apache.cassandra.simulator.cluster.ClusterActions.Options.noActions;


/**
 * In order to run these tests in your IDE, you need to first build a simulator jara
 *
 *    ant simulator-jars
 *
 * And then run your test using the following settings (omit add-* if you are running on jdk8):
 *
        -Dstorage-config=$MODULE_DIR$/test/conf
        -Djava.awt.headless=true
        -javaagent:$MODULE_DIR$/lib/jamm-0.4.0.jar
        -ea
        -Dcassandra.debugrefcount=true
        -Xss384k
        -XX:SoftRefLRUPolicyMSPerMB=0
        -XX:ActiveProcessorCount=2
        -XX:HeapDumpPath=build/test
        -Dcassandra.test.driver.connection_timeout_ms=10000
        -Dcassandra.test.driver.read_timeout_ms=24000
        -Dcassandra.memtable_row_overhead_computation_step=100
        -Dcassandra.test.use_prepared=true
        -Dcassandra.test.sstableformatdevelopment=true
        -Djava.security.egd=file:/dev/urandom
        -Dcassandra.testtag=.jdk11
        -Dcassandra.keepBriefBrief=true
        -Dcassandra.allow_simplestrategy=true
        -Dcassandra.strict.runtime.checks=true
        -Dcassandra.reads.thresholds.coordinator.defensive_checks_enabled=true
        -Dcassandra.test.flush_local_schema_changes=false
        -Dcassandra.test.messagingService.nonGracefulShutdown=true
        -Dcassandra.use_nix_recursive_delete=true
        -Dcie-cassandra.disable_schema_drop_log=true
        -Dlogback.configurationFile=file://$MODULE_DIR$/test/conf/logback-simulator.xml
        -Dcassandra.ring_delay_ms=10000
        -Dcassandra.tolerate_sstable_size=true
        -Dcassandra.skip_sync=true
        -Dcassandra.debugrefcount=false
        -Dcassandra.test.simulator.determinismcheck=strict
        -Dcassandra.test.simulator.print_asm=none
        -javaagent:$MODULE_DIR$/build/test/lib/jars/simulator-asm.jar
        -Xbootclasspath/a:$MODULE_DIR$/build/test/lib/jars/simulator-bootstrap.jar
        -XX:ActiveProcessorCount=4
        -XX:-TieredCompilation
        -XX:-BackgroundCompilation
        -XX:CICompilerCount=1
        -XX:Tier4CompileThreshold=1000
        -XX:ReservedCodeCacheSize=256M
        -Xmx16G
        -Xmx4G
        --add-exports java.base/jdk.internal.misc=ALL-UNNAMED
        --add-exports java.base/jdk.internal.ref=ALL-UNNAMED
        --add-exports java.base/sun.nio.ch=ALL-UNNAMED
        --add-exports java.management.rmi/com.sun.jmx.remote.internal.rmi=ALL-UNNAMED
        --add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED
        --add-exports java.rmi/sun.rmi.server=ALL-UNNAMED
        --add-exports java.sql/java.sql=ALL-UNNAMED
        --add-exports java.rmi/sun.rmi.registry=ALL-UNNAMED
        --add-opens java.base/java.lang.module=ALL-UNNAMED
        --add-opens java.base/java.net=ALL-UNNAMED
        --add-opens java.base/jdk.internal.loader=ALL-UNNAMED
        --add-opens java.base/jdk.internal.ref=ALL-UNNAMED
        --add-opens java.base/jdk.internal.reflect=ALL-UNNAMED
        --add-opens java.base/jdk.internal.math=ALL-UNNAMED
        --add-opens java.base/jdk.internal.module=ALL-UNNAMED
        --add-opens java.base/jdk.internal.util.jar=ALL-UNNAMED
        --add-opens jdk.management/com.sun.management.internal=ALL-UNNAMED
        --add-opens jdk.management.jfr/jdk.management.jfr=ALL-UNNAMED
        --add-opens java.desktop/com.sun.beans.introspect=ALL-UNNAMED
 */
@Command(name = "harry", description = "Harry simulation test", mixinStandardHelpOptions = true)
public class HarrySimulatorTest implements Runnable
{
    private static final Logger logger = LoggerFactory.getLogger(HarrySimulatorTest.class);

    @Option(names = { "-r", "--rows-per-phase" }, description = "Number of rows to check at each phase of the test")
    public int rowsPerPhase = 10;
    @Option(names = { "--nodes-per-dc" }, description = "How many nodes per dc for replication")
    public int nodesPerDc = 3;
    @Option(names = { "-s", "--seed" }, description = "What seed to run with; in hex format... example: 0x190e6ff01d6")
    public String seed = null;

    public static void main(String... args) throws Throwable
    {
        CommandLine commandLine = new CommandLine(HarrySimulatorTest.class);
        commandLine.execute(args);
    }

    @Override
    public void run()
    {
        try
        {
            testInternal();
        }
        catch (Throwable t)
        {
            logger.error("Error running simulation", t);
            throw new RuntimeException(t);
        }
    }

    @Test
    public void test() throws Exception
    {
        // To rerun a failing test for a given seed, uncomment the below and set the seed
//        this.seed = "<your seed here>";
        testInternal();
    }

    protected void testInternal() throws Exception
    {
        int bootstrapNode1 = 4;
        int bootstrapNode2 = 8;
        int bootstrapNode3 = 12;

        StringBuilder rfString = new StringBuilder();
        Map<String, Integer> rfMap = new HashMap<>();
        for (int i = 0; i < 3; i++)
        {
            String dc = "dc" + i;
            rfMap.put(dc, 3);
            if (i > 0)
                rfString.append(", ");
            rfString.append("'").append(dc).append("'").append(" : ").append(nodesPerDc);
        }

        TokenPlacementModel.NtsReplicationFactor rf = new TokenPlacementModel.NtsReplicationFactor(rfMap);

        ConsistencyLevel cl = ALL;

        simulate((config) -> config
                             .failures(new HaltOnError())
                             .threadCount(1000)
                             .readTimeoutNanos(SECONDS.toNanos(5))
                             .writeTimeoutNanos(SECONDS.toNanos(5))
                             .readTimeoutNanos(SECONDS.toNanos(10))
                             .nodes(12, 12)
                             .dcs(3, 3),
                 (config) -> config.set("cms_default_max_retries", 100)
                                   .set("request_timeout", "10000ms")
                                   .set("progress_barrier_min_consistency_level", ALL)
                                   .set("progress_barrier_default_consistency_level", ALL)
                                   .set("progress_barrier_timeout", "600000ms")
                                   // Backoff should be larger than read timeout, since otherwise we will simply saturate the stage with retries
                                   .set("progress_barrier_backoff", "1000ms")
                                   .set("cms_await_timeout", "600000ms"),
                 arr(),
                 (simulation) -> {
                     simulation.cluster.stream().forEach((IInvokableInstance i) -> {
                         simulation.simulated.failureDetector.markUp(i.config().broadcastAddress());
                     });

                     List<ActionSchedule.Work> work = new ArrayList<>();
                     work.add(work("Set up", run(() -> {
                         for (Map.Entry<String, List<TokenPlacementModel.Node>> e : simulation.nodeState.nodesByDc.entrySet())
                         {
                             List<TokenPlacementModel.Node> nodesInDc = e.getValue();
                             for (int i = 0; i < 3; i++)
                             {
                                 TokenPlacementModel.Node node = nodesInDc.get(i);
                                 simulation.nodeState.unsafeBootstrap(node);
                             }
                         }
                     })));
                     work.add(work("Initial configuration",
                                   lazy(() -> simulation.clusterActions.initializeCluster(new ClusterActions.InitialConfiguration(simulation.nodeState.joined(), new int[0])))));

                     work.add(work("Reconfigure CMS",
                                   reconfigureCMS(simulation.simulated, simulation.cluster, 2, true)));
                     work.add(work("Create Keyspace",
                                   simulation.clusterActions.schemaChange(1,
                                                                          String.format("CREATE KEYSPACE %s WITH replication = {'class': 'NetworkTopologyStrategy', " + rfString + "};",
                                                                                        simulation.schema.keyspace))));
                     work.add(work("Create table",
                                   simulation.clusterActions.schemaChange(1,
                                                                          simulation.schema.compile())));
                     simulation.cluster.stream().forEach(i -> {
                         work.add(work("Output epoch",
                                       lazy(simulation.simulated, i, () -> logger.warn(ClusterMetadata.current().epoch.toString()))));
                     });

                     work.add(interleave("Start generating", HarrySimulatorTest.generateWrites(rowsPerPhase, simulation, cl)));
                     work.add(work("Validate all data locally",
                                   lazy(() -> validateAllLocal(simulation, simulation.nodeState.ring, rf, validateQueryConsistency(), simulation.rng))));

                     return arr(work.toArray(new ActionSchedule.Work[0]));
                 },
                 (simulation) -> {
                     List<ActionSchedule.Work> work = new ArrayList<>();
                     List<Integer> registeredNodes = new ArrayList<>(Arrays.asList(bootstrapNode1, bootstrapNode2, bootstrapNode3));
                     List<Integer> bootstrappedNodes = new ArrayList<>();
                     while (!registeredNodes.isEmpty() || !bootstrappedNodes.isEmpty())
                     {
                         boolean shouldBootstrap = simulation.simulated.random.decide(0.5f);
                         if (shouldBootstrap && registeredNodes.isEmpty())
                             shouldBootstrap = false;
                         if (!shouldBootstrap && bootstrappedNodes.isEmpty())
                             shouldBootstrap = true;

                         int node;
                         if (shouldBootstrap)
                         {
                             node = registeredNodes.remove(0);
                             long token = simulation.simulated.random.uniform(Long.MIN_VALUE, Long.MAX_VALUE);
                             work.add(interleave("Bootstrap and generate data",
                                                 ActionList.of(bootstrap(simulation.simulated, simulation.cluster, token, node)),
                                                 generateWrites(rowsPerPhase, simulation, cl)
                             ));
                             simulation.cluster.stream().forEach(i -> {
                                 work.add(work("Output epoch",
                                               lazy(simulation.simulated, i, () -> logger.warn(ClusterMetadata.current().epoch.toString()))));
                             });
                             work.add(work("Bootstrap",
                                           run(() -> simulation.nodeState.bootstrap(node, token))));
                             work.add(work("Check node state",
                                           assertNodeState(simulation.simulated, simulation.cluster, node, NodeState.JOINED)));
                             bootstrappedNodes.add(node);
                         }
                         else
                         {
                             assert !bootstrappedNodes.isEmpty();
                             node = bootstrappedNodes.remove(0);
                             work.add(interleave("Decommission and generate data",
                                                 ActionList.of(decommission(simulation.simulated, simulation.cluster, node)),
                                                 generateWrites(rowsPerPhase, simulation, cl)
                             ));
                             simulation.cluster.stream().forEach(i -> {
                                 work.add(work("Output epoch",
                                               lazy(simulation.simulated, i, () -> logger.warn(ClusterMetadata.current().epoch.toString()))));
                             });
                             work.add(work("Decommission",
                                           run(() -> simulation.nodeState.decommission(node))));
                             work.add(work("Check node state", assertNodeState(simulation.simulated, simulation.cluster, node, NodeState.LEFT)));
                         }
                         work.add(work("Validate data with " + validateQueryConsistency(),
                                       lazy(() -> validateAllLocal(simulation, simulation.nodeState.ring, rf, validateQueryConsistency(), simulation.rng))));
                         boolean tmp = shouldBootstrap;
                         work.add(work("Output message",
                                       run(() -> logger.warn("Finished {} of {} and data validation!\n", tmp ? "bootstrap" : "decommission", node))));
                     }
                     work.add(work("Output message",
                                   run(() -> logger.warn("Finished!"))));

                     return arr(work.toArray(new ActionSchedule.Work[0]));
                 }
        );
    }

    /**
     * Harry simulation. Assumes no scheduler delays and has ideal networking conditions.
     *
     * Since this is generally used to test _different_ subsystems, we plug in failures into the component
     * we are testing to both reduce the noise and the surface for potential investigations. This also
     * has a nice side effect of making simulations slightly faster.
     */
    static class HarrySimulation implements Simulation
    {
        protected final ClusterActions clusterActions;
        protected final SimulatedNodeState nodeState;

        protected final SimulatedSystems simulated;
        protected final RunnableActionScheduler scheduler;
        protected final Cluster cluster;
        protected final Function<HarrySimulation, ActionSchedule.Work[]> schedule;

        protected final EntropySource rng;
        protected final SchemaSpec schema;
        protected final Generator<OperationsGenerators.ToOp> insertGen;
        protected final QueryBuildingVisitExecutor queryBuilder;
        protected final QuiescentChecker model;

        protected final Map<Long, Visit> log;
        protected final Generator<Long> ltsGen;
        protected final DataTracker tracker;

        private HarrySimulation(SchemaSpec schema,
                                EntropySource rng,
                                SimulatedSystems simulated,
                                RunnableActionScheduler scheduler,
                                Cluster cluster,
                                Function<HarrySimulation, SimulatedNodeState> nodeState,
                                Function<HarrySimulation, ActionSchedule.Work[]> schedule,
                                Map<Long, Visit> log,
                                Generator<Long> ltsGen,
                                DataTracker tracker)
        {
            this.rng = rng;
            this.schema = schema;
            this.insertGen = OperationsGenerators.writeOp(schema);
            this.queryBuilder = new QueryBuildingVisitExecutor(schema, QueryBuildingVisitExecutor.WrapQueries.UNLOGGED_BATCH);
            this.model = new QuiescentChecker(schema.valueGenerators, tracker, new Model.Replay()
            {
                @Override
                public Visit replay(long lts)
                {
                    return log.get(lts);
                }

                @Override
                public Operations.Operation replay(long lts, int opId)
                {
                    return log.get(lts).operations[opId];
                }

                @Override
                public Iterator<Visit> iterator()
                {
                    List<Long> visited = new ArrayList<>(log.keySet());
                    visited.sort(Long::compare);
                    return new Iterator<>()
                    {
                        int idx = 0;

                        @Override
                        public boolean hasNext()
                        {
                            return idx < visited.size();
                        }

                        @Override
                        public Visit next()
                        {
                            return replay(visited.get(idx++));
                        }
                    };
                }
            });

            this.simulated = simulated;
            this.scheduler = scheduler;
            this.cluster = cluster;

            Options options = noActions(cluster.size());
            this.clusterActions = new ClusterActions(simulated, cluster,
                                                     options, new NoOpListener(), new Debug(new EnumMap<>(Debug.Info.class), new int[0]));

            this.nodeState = nodeState.apply(this);
            this.schedule = schedule;

            this.log = log;
            this.ltsGen = ltsGen;
            this.tracker = tracker;
        }

        public HarrySimulation withScheduler(RunnableActionScheduler scheduler)
        {
            return new HarrySimulation(schema, rng, simulated, scheduler, cluster, (ignore) -> nodeState, schedule, log, ltsGen, tracker);
        }

        public HarrySimulation withSchedulers(Function<HarrySimulation, Map<Verb, FutureActionScheduler>> schedulers)
        {
            Map<Verb, FutureActionScheduler> perVerbFutureActionScheduler = schedulers.apply(this);
            SimulatedSystems simulated = new SimulatedSystems(this.simulated.random,
                                                              this.simulated.time,
                                                              this.simulated.delivery,
                                                              this.simulated.execution,
                                                              this.simulated.ballots,
                                                              this.simulated.failureDetector,
                                                              this.simulated.snitch,
                                                              this.simulated.futureScheduler,
                                                              perVerbFutureActionScheduler,
                                                              this.simulated.debug,
                                                              this.simulated.failures);
            return new HarrySimulation(schema, rng, simulated, scheduler, cluster, (ignore) -> nodeState, schedule, log, ltsGen, tracker);
        }

        public HarrySimulation withSchedule(Function<HarrySimulation, ActionSchedule.Work[]> schedule)
        {
            return new HarrySimulation(schema, rng, simulated, scheduler, cluster, (ignore) -> nodeState, schedule, log, ltsGen, tracker);
        }

        @Override
        public CloseableIterator<?> iterator()
        {
            return new ActionSchedule(simulated.time, simulated.futureScheduler, () -> 0L, scheduler, schedule.apply(this));
        }

        @Override
        public void run()
        {
            try (CloseableIterator<?> iter = iterator())
            {
                while (iter.hasNext())
                {
                    checkForErrors();
                    iter.next();
                }
                checkForErrors();
            }
        }

        private void checkForErrors()
        {
            if (simulated.failures.hasFailure())
            {
                AssertionError error = new AssertionError("Errors detected during simulation");
                // don't care about the stack trace... the issue is the errors found and not what part of the scheduler we stopped
                error.setStackTrace(new StackTraceElement[0]);
                simulated.failures.get().forEach(error::addSuppressed);
                throw error;
            }
        }

        public void close() throws Exception
        {
        }
    }

    class HarrySimulationBuilder extends ClusterSimulation.Builder<HarrySimulation>
    {
        protected final Consumer<IInstanceConfig> configUpdater;

        HarrySimulationBuilder(Consumer<IInstanceConfig> configUpdater)
        {
            this.configUpdater = configUpdater;
        }

        @Override
        public Map<Verb, FutureActionScheduler> perVerbFutureActionSchedulers(int nodeCount, SimulatedTime time, RandomSource random, FutureActionScheduler defaultScheduler)
        {
            return networkSchedulers(nodeCount, time, random);
        }

        @Override
        public FutureActionScheduler futureActionScheduler(int nodeCount, SimulatedTime time, RandomSource random)
        {
            return new AlwaysDeliverNetworkScheduler(time);
        }

        @Override
        public ClusterSimulation<HarrySimulation> create(long seed) throws IOException
        {
            RandomSource random = new RandomSource.Default();
            random.reset(seed);

            return new ClusterSimulation<>(random, seed, 1, this, configUpdater,
                                           (simulated, scheduler, cluster, options) -> {
                                               EntropySource rng = new JdkRandomEntropySource(seed);
                                               SchemaSpec schema = schemaSpecGen("harry", "tbl").generate(rng);
                                               return new HarrySimulation(schema,
                                                                          rng,
                                                                          simulated,
                                                                          scheduler,
                                                                          cluster,
                                                                          SimulatedNodeState::new,
                                                                          // No work initially
                                                                          (sim) -> new ActionSchedule.Work[0],
                                                                          new HashMap<>(),
                                                                          OperationsGenerators.lts(),
                                                                          new DataTracker.SimpleDataTracker());
                                           });
        }
    }

    /**
     * Simulation entrypoint; syntax sugar for creating a simulation.
     */
    void simulate(Consumer<ClusterSimulation.Builder<HarrySimulation>> configure,
                  Consumer<IInstanceConfig> instanceConfigUpdater,
                  String[] properties,
                  Function<HarrySimulation, ActionSchedule.Work[]>... phases) throws IOException
    {
        try (WithProperties p = new WithProperties().with(properties))
        {
            HarrySimulationBuilder factory = new HarrySimulationBuilder(instanceConfigUpdater);

            SimulationRunner.beforeAll();
            long seed = SimulationRunner.parseHex(Optional.ofNullable(this.seed)).orElseGet(() -> new Random().nextLong());
            logger.info("Seed 0x{}", Long.toHexString(seed));
            configure.accept(factory);
            try (ClusterSimulation<HarrySimulation> clusterSimulation = factory.create(seed))
            {
                HarrySimulation simulation = clusterSimulation.simulation();

                // For better determinism during startup, we allow instances to fully start (including daemon work)
                for (int i = 0; i < phases.length; i++)
                {
                    HarrySimulation current = simulation;
                    if (i == 0)
                        current = current.withScheduler(new RunnableActionScheduler.Immediate()).withSchedulers((s) -> Collections.emptyMap());
                    current.withSchedule(phases[i]).run();
                }
            }
            catch (Throwable t)
            {
                if (t instanceof SimulationException) throw t;
                throw new SimulationException(seed, "Failure creating the simulation", t);
            }
        }
    }

    /**
     * Custom network scheduler for testing TCM.
     */
    public Map<Verb, FutureActionScheduler> networkSchedulers(int nodes, SimulatedTime time, RandomSource random)
    {
        Set<Verb> extremelyLossy = new HashSet<>(Arrays.asList(Verb.TCM_ABORT_MIG, Verb.TCM_REPLICATION,
                                                               Verb.TCM_COMMIT_REQ, Verb.TCM_NOTIFY_REQ,
                                                               Verb.TCM_INIT_MIG_REQ, Verb.TCM_INIT_MIG_RSP,
                                                               Verb.TCM_DISCOVER_REQ, Verb.TCM_DISCOVER_RSP));

        Set<Verb> somewhatSlow = new HashSet<>(Arrays.asList(Verb.BATCH_STORE_REQ, Verb.BATCH_STORE_RSP));

        Set<Verb> somewhatLossy = new HashSet<>(Arrays.asList(Verb.TCM_CURRENT_EPOCH_REQ,
                                                              Verb.TCM_NOTIFY_RSP, Verb.TCM_FETCH_CMS_LOG_RSP,
                                                              Verb.TCM_FETCH_PEER_LOG_RSP, Verb.TCM_COMMIT_RSP,
                                                              Verb.TCM_FETCH_CMS_LOG_REQ, Verb.TCM_FETCH_PEER_LOG_REQ,
                                                              Verb.PAXOS2_COMMIT_REMOTE_REQ, Verb.PAXOS2_COMMIT_REMOTE_RSP,
                                                              Verb.PAXOS2_PREPARE_REQ, Verb.PAXOS2_PREPARE_RSP,
                                                              Verb.PAXOS2_PROPOSE_REQ, Verb.PAXOS2_PROPOSE_RSP,
                                                              Verb.PAXOS_PREPARE_REQ, Verb.PAXOS_PREPARE_RSP,
                                                              Verb.PAXOS_PROPOSE_RSP, Verb.PAXOS_PROPOSE_REQ,
                                                              Verb.PAXOS_COMMIT_REQ, Verb.PAXOS_COMMIT_RSP));

        Map<Verb, FutureActionScheduler> schedulers = new HashMap<>();
        for (Verb verb : Verb.values())
        {
            if (extremelyLossy.contains(verb))
                schedulers.put(verb, new FixedLossNetworkScheduler(nodes, random, time, KindOfSequence.UNIFORM, .15f, .20f));
            else if (somewhatLossy.contains(verb))
                schedulers.put(verb, new FixedLossNetworkScheduler(nodes, random, time, KindOfSequence.UNIFORM, .1f, .15f));
            else if (somewhatSlow.contains(verb))
                schedulers.put(verb, new AlwaysDeliverNetworkScheduler(time, TimeUnit.MILLISECONDS.toNanos(100)));
        }
        return schedulers;
    }

    public static Action reconfigureCMS(SimulatedSystems simulated, Cluster cluster, int rf, boolean inEachDc)
    {
        return new SimulatedActionTask("", Action.Modifiers.RELIABLE_NO_TIMEOUTS, Action.Modifiers.RELIABLE_NO_TIMEOUTS, null, simulated,
                                       new InterceptedExecution.InterceptedRunnableExecution((InterceptingExecutor) cluster.get(1).executor(),
                                                                                             cluster.get(1).transfer((IIsolatedExecutor.SerializableRunnable) () -> {
                                                                                                 ReplicationParams params;
                                                                                                 if (inEachDc)
                                                                                                 {
                                                                                                     Map<String, Integer> rfs = new HashMap<>();
                                                                                                     for (String dc : ClusterMetadata.current().directory.knownDatacenters())
                                                                                                     {
                                                                                                         rfs.put(dc, rf);
                                                                                                     }
                                                                                                     params = ReplicationParams.ntsMeta(rfs);
                                                                                                 }
                                                                                                 else
                                                                                                 {
                                                                                                     params = ReplicationParams.simpleMeta(rf, ClusterMetadata.current().directory.knownDatacenters());
                                                                                                 }
                                                                                                 ClusterMetadataService.instance().reconfigureCMS(params);
                                                                                             })));
    }

    /**
     * Creates an action that is equivalent to starting bootstrap on the node via nodetool bootstrap, or
     * simply starting a fresh node with a pre-configured token.
     */
    public static Action bootstrap(SimulatedSystems simulated, Cluster cluster, long token, int node)
    {
        IIsolatedExecutor.SerializableRunnable runnable = () -> {
            try
            {
                Startup.startup(() -> new PrepareJoin(ClusterMetadata.current().myNodeId(),
                                                      Collections.singleton(new Murmur3Partitioner.LongToken(token)),
                                                      ClusterMetadataService.instance().placementProvider(),
                                                      true,
                                                      true),
                                true,
                                true,
                                false);
            }
            catch (Throwable t)
            {
                logger.error("Could not bootstrap. Interrupting simulation.", t);
                SimulatorUtils.failWithOOM();
            }
        };

        return new SimulatedActionTask("Bootstrap Node",
                                       Action.Modifiers.NONE,
                                       Action.Modifiers.NONE,
                                       null,
                                       simulated,
                                       new InterceptedExecution.InterceptedRunnableExecution((InterceptingExecutor) cluster.get(node).executor(),
                                                                                             cluster.get(node).transfer(runnable)));
    }

    public static Action decommission(SimulatedSystems simulated, Cluster cluster, int node)
    {
        IIsolatedExecutor.SerializableRunnable runnable = () -> {
            try
            {
                SingleNodeSequences.decommission(false, false);
            }
            catch (Throwable t)
            {
                logger.error("Could not decommission. Interrupting simulation.", t);
                SimulatorUtils.failWithOOM();
            }
        };

        return new SimulatedActionTask("Decommission Node",
                                       Action.Modifiers.NONE,
                                       Action.Modifiers.NONE,
                                       null,
                                       simulated,
                                       new InterceptedExecution.InterceptedRunnableExecution((InterceptingExecutor) cluster.get(node).executor(),
                                                                                             cluster.get(node).transfer(runnable)));
    }

    public static Action run(Runnable run)
    {
        return new Actions.LambdaAction("", Action.Modifiers.RELIABLE_NO_TIMEOUTS, () -> {
            run.run();
            return ActionList.empty();
        });
    }

    public static Action lazy(SimulatedSystems simulated, IInvokableInstance instance, IIsolatedExecutor.SerializableRunnable runnable)
    {
        return new SimulatedActionTask("", Action.Modifiers.RELIABLE_NO_TIMEOUTS, Action.Modifiers.RELIABLE_NO_TIMEOUTS, null, simulated,
                                       new InterceptedExecution.InterceptedRunnableExecution((InterceptingExecutor) instance.executor(), instance.transfer(runnable)));
    }

    public static Action lazy(Supplier<Action> run)
    {
        return new Actions.LambdaAction("", Action.Modifiers.RELIABLE_NO_TIMEOUTS, () -> ActionList.of(run.get()));
    }

    private static Action assertNodeState(SimulatedSystems simulated, Cluster cluster, int i, NodeState expected)
    {
        return lazy(simulated, cluster.get(i),
                    () -> {
                        NodeState actual = ClusterMetadata.current().myNodeState();
                        if (!actual.toString().equals(expected.toString()))
                        {
                            logger.error("Node {} state ({}) is not as expected {}", i, actual, expected);
                            SimulatorUtils.failWithOOM();
                        }
                    });
    }

    /**
     * Creates an action list with a fixed number of data-generating operations that conform to the given Harry configuration.
     */
    public static ActionList generateWrites(int ops, HarrySimulation simulation, ConsistencyLevel cl)
    {
        Action[] actions = new Action[ops];
        OrderOn orderOn = new OrderOn.Strict(actions, 2);
        for (int i = 0; i < ops; i++)
        {
            long lts = simulation.ltsGen.generate(simulation.rng);

            Visit visit = new Visit(lts, new Operations.Operation[]{ simulation.insertGen.generate(simulation.rng).toOp(lts) });
            Visit prev_ = simulation.log.put(lts, visit);
            Invariants.require(prev_ == null);

            actions[i] = new Actions.LambdaAction("", Action.Modifiers.RELIABLE_NO_TIMEOUTS, () -> {
                CompiledStatement compiledStatement = simulation.queryBuilder.compile(visit);
                DataTracker tracker = simulation.tracker;

                int[] joined = simulation.nodeState.joined();
                int coordinator = joined[simulation.rng.nextInt(joined.length)];
                RetryingQuery query = new RetryingQuery(compiledStatement.cql(), cl, compiledStatement.bindings());
                Action wrapper = new SimulatedActionCallable<>("Query",
                                                               Action.Modifiers.RELIABLE_NO_TIMEOUTS,
                                                               Action.Modifiers.RELIABLE_NO_TIMEOUTS,
                                                               simulation.simulated,
                                                               simulation.cluster.get(coordinator),
                                                               query)
                {
                    @Override
                    protected InterceptedExecution.InterceptedTaskExecution task()
                    {
                        return new InterceptedExecution.InterceptedTaskExecution((InterceptingExecutor) on.executor())
                        {
                            public void run()
                            {
                                tracker.begin(visit);
                                // we'll be invoked on the node's executor, but we need to ensure the task is loaded on its classloader
                                try
                                {
                                    accept(on.unsafeCallOnThisThread(execute), null);
                                }
                                catch (Throwable t)
                                {
                                    accept(null, t);
                                }
                                finally
                                {
                                    execute = null;
                                }
                            }
                        };
                    }

                    @Override
                    public void accept(SimpleQueryResult result, Throwable failure)
                    {
                        if (failure != null)
                            simulated.failures.accept(failure);
                        else
                            tracker.end(visit);
                    }
                };

                return ActionList.of(wrapper);
            });
        }
        return ActionList.of(actions).orderOn(orderOn);
    }

    public static class RetryingQuery extends Query
    {
        public RetryingQuery(String query, ConsistencyLevel cl, Object[] boundValues)
        {
            super(query, -1, cl, null, boundValues);
        }

        @Override
        public SimpleQueryResult call()
        {
            while (true)
            {
                try
                {
                    return super.call();
                }
                catch (UncheckedInterruptedException e)
                {
                    throw new RuntimeException(e);
                }
                catch (Throwable t)
                {
                    logger.error("Caught error while executing query. Will ignore and retry: " + t.getMessage());
                }
            }
        }
    }

    /**
     * Given you have used `generate` methods to generate data with Harry, you can use this method to check whether all
     * data has been propagated everywhere it should be, be it via streaming, read repairs, or regular writes.
     */
    public static Action validateAllLocal(HarrySimulation simulation, List<TokenPlacementModel.Node> owernship, TokenPlacementModel.ReplicationFactor rf, ConsistencyLevel consistencyLevel, EntropySource rng)
    {
        return new Actions.LambdaAction("Validate", Action.Modifiers.RELIABLE_NO_TIMEOUTS,
                                        () -> {
                                            if (!simulation.tracker.allFinished())
                                                throw new IllegalStateException("Can not begin validation, as writing has not quiesced yet: " + simulation.tracker);

                                            logger.warn("Starting validation. Ring view: {}", simulation.nodeState);
                                            Set<Long> pds = visitedPds(simulation);
                                            List<Action> actions = new ArrayList<>();
                                            for (Long pd : pds)
                                            {
                                                Operations.SelectPartition select = new Operations.SelectPartition(Long.MAX_VALUE, pd);
                                                actions.add(new HarryValidatingQuery(simulation, simulation.cluster, rf,
                                                                                     owernship, new Visit(Long.MAX_VALUE, new Operations.Operation[]{ select }),
                                                                                     simulation.queryBuilder,
                                                                                     consistencyLevel,
                                                                                     rng));
                                            }
                                            return ActionList.of(actions).setStrictlySequential();
                                        });
    }

    protected ConsistencyLevel validateQueryConsistency()
    {
        return ConsistencyLevel.NODE_LOCAL;
    }

    private static Set<Long> visitedPds(HarrySimulation simulation)
    {
        Set<Long> pds = new HashSet<>();
        for (Visit visit : simulation.log.values())
        {
            for (Operations.Operation operation : visit.operations)
            {
                if (operation instanceof Operations.PartitionOperation)
                    pds.add(((Operations.PartitionOperation) operation).pd);
            }
        }
        return pds;
    }

    private static ActionSchedule.Work work(String toString, Action... actions)
    {
        return new ActionSchedule.Work(UNLIMITED, Collections.singletonList(ActionList.of(actions).setStrictlySequential())) {
            @Override
            public String toString()
            {
                return toString;
            }
        };
    }

    private static ActionSchedule.Work interleave(String toString, ActionList... actions)
    {
        return new ActionSchedule.Work(UNLIMITED, Arrays.asList(actions)) {
            @Override
            public String toString()
            {
                return toString;
            }
        };
    }

    /**
     * Simple simulated node state. Used to closely track what is going on in the cluster and
     * model placements for node-local validation.
     */
    public static class SimulatedNodeState
    {
        public final TokenPlacementModel.Node[] nodesLookup;
        public final List<TokenPlacementModel.Node> ring;
        public final Map<String, List<TokenPlacementModel.Node>> nodesByDc;
        public final Map<String, Integer> idByAddr;

        public SimulatedNodeState(HarrySimulation simulation)
        {
            this.nodesLookup = new TokenPlacementModel.Node[simulation.cluster.size()];
            this.nodesByDc = new HashMap<>();
            this.idByAddr = new HashMap<>();
            for (int i = 0; i < simulation.cluster.size(); i++)
            {
                int nodeId = i + 1;
                IInstanceConfig config = simulation.cluster.get(nodeId).config();

                InetAddressAndPort addr = InetAddressAndPort.getByAddress(config.broadcastAddress());

                TokenPlacementModel.Node node = new TokenPlacementModel.Node(0, 0, 0, 0,
                                                                             constantLookup(addr.toString(),
                                                                                            Long.parseLong(config.getString("initial_token")),
                                                                                            simulation.clusterActions.snitch.get().getDatacenter(addr),
                                                                                            simulation.clusterActions.snitch.get().getRack(addr)));
                nodesLookup[i] = node;
                nodesByDc.computeIfAbsent(node.dc(), (k) -> new ArrayList<>()).add(node);
                idByAddr.put(addr.toString(), config.num());
            }
            this.ring = new ArrayList<>();
        }

        public int[] joined()
        {
            int[] joined = new int[ring.size()];
            for (int i = 0; i < ring.size(); i++)
            {
                joined[i] = idByAddr.get(ring.get(i).id());
            }
            return joined;
        }

        public void unsafeBootstrap(int nodeId)
        {
            TokenPlacementModel.Node n = nodesLookup[nodeId - 1];
            int idx = Collections.binarySearch(ring, n);
            if (idx < 0)
                ring.add(-idx - 1, n);
            else
                ring.set(idx, n);
        }

        public void unsafeBootstrap(TokenPlacementModel.Node n)
        {
            int idx = Collections.binarySearch(ring, n);
            if (idx < 0)
                ring.add(-idx - 1, n);
            else
                ring.set(idx, n);
        }

        public void bootstrap(int nodeId, long token)
        {
            System.out.printf("Marking %d as bootstrapped%n", nodeId);
            TokenPlacementModel.Node n = nodesLookup[nodeId - 1];
            // Update token in the lookup. Do it before search as we need ring sorted
            n = withToken(n, token);
            nodesLookup[nodeId - 1] = n;

            int idx = Collections.binarySearch(ring, n);

            if (idx < 0)
                ring.add(-idx - 1, n);
            else
                ring.set(idx, n);

            // Assert sorted
            TokenPlacementModel.Node prev = null;
            for (TokenPlacementModel.Node node : ring)
            {
                if (prev != null)
                    assert node.token() > prev.token() : "Ring doesn't seem to be sorted: " + ring;
                prev = node;
            }
        }

        public void decommission(int nodeId)
        {
            System.out.printf("Marking %d as decommissioned%n", nodeId);
            TokenPlacementModel.Node n = nodesLookup[nodeId - 1];
            int idx = Collections.binarySearch(ring, n);
            ring.remove(idx);
            assertSorted(ring);
        }

        private static void assertSorted(List<TokenPlacementModel.Node> ring)
        {
            // Assert sorted
            TokenPlacementModel.Node prev = null;
            for (TokenPlacementModel.Node node : ring)
            {
                if (prev != null)
                    assert node.token() > prev.token() : "Ring doesn't seem to be sorted: " + ring;
                prev = node;
            }

        }
        private static TokenPlacementModel.Node withToken(TokenPlacementModel.Node node, long token)
        {
            return new TokenPlacementModel.Node(0, 0, 0, 0,
                                                constantLookup(node.id(),
                                                               token,
                                                               node.dc(),
                                                               node.rack()));
        }

        public String toString()
        {
            return "SimulatedNodeState{" +
                   "ring=" + ring +
                   '}';
        }
    }

    public static <T> T[] arr(T... arr)
    {
        return arr;
    }

    public Generator<SchemaSpec> schemaSpecGen(String keyspace, String prefix)
    {
        return SchemaGenerators.schemaSpecGen(keyspace, prefix, 1000);
    }

    public static class HaltOnError extends Failures
    {
        @Override
        public void onFailure(Throwable t)
        {
            super.onFailure(t);
            logger.error("Caught an exception, going to halt", t);
        }
    }
}
