content: /*
   * 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 accord.coordinate;
  
  import java.util.List;
  import java.util.function.BiConsumer;
  
  import javax.annotation.Nullable;
  
  import accord.api.Data;
  import accord.api.Result;
  import accord.api.Timeouts;
  import accord.coordinate.ExecuteFlag.CoordinationFlags;
  import accord.coordinate.ExecuteFlag.ExecuteFlags;
  import accord.coordinate.tracking.QuorumIdTracker;
  import accord.coordinate.tracking.RequestStatus;
  import accord.local.Commands;
  import accord.local.Commands.CommitOutcome;
  import accord.local.LogUnavailableException;
  import accord.local.Node;
  import accord.local.Node.Id;
  import accord.local.SafeCommand;
  import accord.local.SafeCommandStore;
  import accord.local.SequentialAsyncExecutor;
  import accord.local.StoreParticipants;
  import accord.messages.Accept;
  import accord.messages.Callback;
  import accord.messages.Commit;
  import accord.messages.InformDurable;
  import accord.messages.MessageType;
  import accord.messages.ReadData;
  import accord.messages.ReadData.CommitOrReadNack;
  import accord.messages.ReadData.ReadOk;
  import accord.messages.ReadData.ReadReply;
  import accord.messages.ReadTxnData;
  import accord.messages.SafeCallback;
  import accord.messages.StableThenRead;
  import accord.primitives.Ballot;
  import accord.primitives.Deps;
  import accord.primitives.FullRoute;
  import accord.primitives.Participants;
  import accord.primitives.Ranges;
  import accord.primitives.Timestamp;
  import accord.primitives.TimestampWithUniqueHlc;
  import accord.primitives.Txn;
  import accord.primitives.TxnId;
  import accord.primitives.Unseekables;
  import accord.primitives.Writes;
  import accord.topology.Topologies;
  import accord.utils.Invariants;
  import accord.utils.SortedArrays;
  import accord.utils.SortedListSet;
  import accord.utils.UnhandledEnum;
  import org.agrona.collections.IntHashSet;
  
  import static accord.api.ProtocolModifiers.Toggles.fastReadExecMayResendTxn;
  import static accord.api.ProtocolModifiers.Toggles.fastReadsMayBypassSafeStore;
  import static accord.api.ProtocolModifiers.Toggles.permitLocalExecution;
  import static accord.api.ProtocolModifiers.Toggles.sendNoStableIfFastExec;
  import static accord.api.ProtocolModifiers.Toggles.sendOnlyReadStableMessages;
  import static accord.coordinate.CoordinationAdapter.Factory.Kind.Standard;
  import static accord.coordinate.ExecuteFlag.READY_TO_EXECUTE;
  import static accord.coordinate.ExecutePath.FAST;
  import static accord.coordinate.ExecutePath.RECOVER;
  import static accord.coordinate.ReadCoordinator.Action.Approve;
  import static accord.coordinate.ReadCoordinator.Action.ApprovePartial;
  import static accord.messages.Commit.Kind.StableFastPath;
  import static accord.messages.Commit.Kind.StableMediumPath;
  import static accord.messages.Commit.Kind.StableSlowPath;
  import static accord.messages.Commit.Kind.StableWithTxnAndDeps;
  import static accord.messages.ReadData.CommitOrReadNack.Waiting;
  import static accord.primitives.SaveStatus.Stable;
  import static accord.primitives.Status.Durability.DurablyStable;
  import static accord.primitives.Status.Phase.Execute;
  import static java.util.concurrent.TimeUnit.MICROSECONDS;
  
  // TODO (expected): return Waiting from ReadData if not ready to execute, and do not submit more than one speculative retry in this case
  // TODO (expected): by default, if we can execute locally, never contact a remote replica regardless of local outcome
  public class ExecuteTxn extends ReadCoordinator<Result, ReadReply>
  {
      class StableTracker extends QuorumIdTracker implements Callback<ReadReply>
      {
          private boolean isDone;
          private boolean informOnSuccess;
  
          public StableTracker(Topologies topologies)
          {
              super(topologies);
          }
  
          @Override
          public void onSuccess(Id from, ReadReply reply)
          {
              if ((reply.isOk() || reply == Waiting) && RequestStatus.Success == recordSuccess(from) && informOnSuccess)
                  informStable();
          }
  
          @Override
          public void onFailure(Id from, Throwable failure)
          {
          }
  
          void informStable()
          {
              Invariants.require(hasReachedQuorum());
              isDone = true;
              InformDurable.informHome(node, topologies, txnId, route, executeAt, DurablyStable);
          }
  
          void maybeInformStable()
          {
              if (!isDone && stable.hasReachedQuorum())
                  informStable();
          }
  
          void setDone()
          {
              isDone = true;
          }
  
          void informStableOnceQuorum()
          {
              if (!isDone)
              {
                  if (stable.hasReachedQuorum()) informStable();
                  else informOnSuccess = true;
              }
          }
      }
  
      final ExecutePath path;
      final Txn txn;
      final FullRoute<?> route;
      final Ballot ballot;
      final Timestamp executeAt;
      final Deps stableDeps;
      final Deps sendDeps;
      final Topologies allTopologies;
      final CoordinationFlags flags;
      private final StableTracker stable;
      private @Nullable SortedListSet<Node.Id> unstableFastReads;
  
      private final Participants<?> readScope;
      private Data data;
      private long uniqueHlc;
      private boolean isPrivilegedVoteCommitting;
  
      ExecuteTxn(Node node, SequentialAsyncExecutor executor, Topologies topologies, FullRoute<?> route, Ballot ballot, ExecutePath path, CoordinationFlags flags, TxnId txnId, Txn txn, Timestamp executeAt, Deps stableDeps, Deps sendDeps, BiConsumer<? super Result, Throwable> callback)
      {
          super(node, executor, topologies.forEpoch(executeAt.epoch()), txnId, callback);
          this.path = ballot == Ballot.ZERO ? path : RECOVER;
          this.txn = txn;
          this.route = route;
          this.ballot = ballot;
          this.allTopologies = topologies;
          this.executeAt = executeAt;
          this.stableDeps = stableDeps;
          this.sendDeps = sendDeps;
          this.flags = flags;
          this.stable = new StableTracker(topologies.forEpochs(txnId.epoch(), executeAt.epoch()));
          this.readScope = txn == null ? route : route.intersecting(txn.keys());
          Invariants.require(!txnId.awaitsOnlyDeps());
          Invariants.require(!txnId.awaitsPreviouslyOwned());
      }
  
      @Override
      protected void startOnceInitialised()
      {
          node.agent().coordinatorEvents().onExecuting(txnId, ballot, stableDeps, path);
          Node.Id self = node.id();
          if (permitLocalExecution() && tryIfUniversal(self))
          {
              isPrivilegedVoteCommitting = txnId.hasPrivilegedCoordinator() && path == FAST;
              ExecuteFlags executeFlags = flags.get(self);
              /*
                This decides if our local execution may fast execute with the provided flags.
                LocalExecute is treated specially because the privileged coordinator optimisation requires that
                the coordinator record STABLE before sending further messages to any other replica.
                So, if the privileged coordinator optimisation is enabled and we _are_ the privileged coordinator
                taking the fast path, it is unsafe to perform a fast read (that skips updating the Accord state machine)
                because it would be unsafe to continue to the next phase until the local coordinator has updated its state machine.
                It woudl be possible to push this work onto the Persist phase, so that we have an equivalent LocalExecute that
                ensures PREAPPLIED is recorded at the local coordinator before any other replicas are sent Apply, but
                for now we simply disable this optimisation in this case.
               */
              boolean mayFastExecute = !isPrivilegedVoteCommitting
                                       && executeFlags.contains(READY_TO_EXECUTE)
                                       && fastReadsMayBypassSafeStore(txnId);
  
              if (mayFastExecute)
                  markUnstableFastRead(node.id());
              new LocalExecute(txnId, executeFlags, mayFastExecute).process(node, node.agent().selfExpiresAt(txnId, Execute, MICROSECONDS));
          }
          else if (path == FAST && txnId.hasPrivilegedCoordinator())
          {
              // we can't safely take the fast path via PRIVILEGED_COORDINATOR optimisation if we aren't permitted to execute locally,
              // so we take the MEDIUM or SLOW path
              adapter().propose(node, executor, null, route, txnId.hasMediumPath() ? Accept.Kind.MEDIUM : Accept.Kind.SLOW,
                                Ballot.ZERO, txnId, txn, executeAt, stableDeps, takeCallback());
          }
          else
          {
              super.startOnceInitialised();
          }
      }
  
      @Override
      protected void start(List<Id> readFrom)
      {
          // TODO (desired): migrate to SortedListSet; or introduce a specialised version for integer keys; or introduce a hash equivalent
          Topologies all = allTopologies;
          Commit.Kind kind = commitKind();
          for (int i = 0, size = readFrom.size() ; i < size ; ++i)
          {
              Node.Id to = readFrom.get(i);
              ExecuteFlags flags = this.flags.get(to);
              Invariants.require(kind.compareTo(StableFastPath) >= 0);
              boolean sendUnstable = flags.contains(READY_TO_EXECUTE) && sendNoStableIfFastExec() && path != RECOVER;
              if (sendUnstable) sendUnstableRead(to, flags);
              else sendStableRead(to, kind);
          }
  
          if (sendOnlyReadStableMessages() && (all.size() == 1 || all.current().nodes().containsAll(all.nodes())))
              return;
  
          IntHashSet readSet = new IntHashSet();
          readFrom.forEach(i -> readSet.add(i.id));
          SortedArrays.SortedArrayList<Id> contact = all.nodes().without(all::isFaulty);
          for (Node.Id to : contact)
          {
              if (readSet.contains(to.id))
                  continue;
  
              if (sendOnlyReadStableMessages() && all.current().contains(to))
                  continue;
  
              sendStableOnly(to, kind);
          }
      }
  
      private void sendStableOnly(Node.Id to, Commit.Kind kind)
      {
          Commit send = new Commit(kind, to, allTopologies, txnId, txn, route, ballot, executeAt, stableDeps);
          boolean addCallback = allTopologies.size() == 1 || stable.nodes().contains(to);
          if (addCallback) node.send(to, send, executor, stable);
          else node.send(to, send);
  
      }
  
      private void sendUnstableRead(Node.Id to, ExecuteFlags flags)
      {
          Txn sendTxn = null;
          Timestamp sendExecuteAt = null;
          if (flags.contains(READY_TO_EXECUTE) && fastReadExecMayResendTxn() && fastReadsMayBypassSafeStore(txnId))
          {
              sendTxn = txn;
              sendExecuteAt = executeAt;
              markUnstableFastRead(to);
          }
          node.send(to, new ReadTxnData(to, allTopologies, txnId, readScope, sendTxn, sendExecuteAt, executeAt.epoch(), flags), executor, this);
      }
  
      private void markUnstableFastRead(Node.Id to)
      {
          if (unstableFastReads == null)
              unstableFastReads = SortedListSet.noneOf(allTopologies.current().nodes());
          unstableFastReads.add(to);
      }
  
      private void sendStableRead(Node.Id to, Commit.Kind kind)
      {
          node.send(to, new StableThenRead(kind, to, allTopologies, txnId, txn, route, executeAt, stableDeps), executor, this);
      }
  
      private void sendMaximal(Node.Id to)
      {
          Commit send = new Commit(StableWithTxnAndDeps, to, allTopologies, txnId, txn, route, ballot, executeAt, stableDeps);
          node.send(to, send, executor, stable);
      }
  
      private Commit.Kind commitKind()
      {
          switch (path)
          {
              default: throw new UnhandledEnum(path);
              case FAST:    return StableFastPath;
              case MEDIUM:  return StableMediumPath;
              case SLOW:    return StableSlowPath;
              case RECOVER: return StableWithTxnAndDeps;
          }
      }
  
      @Override
      public void contact(Id to)
      {
          ExecuteFlags flags = this.flags.get(to);
          boolean sendUnstable = !sendOnlyReadStableMessages() || path == RECOVER || flags.contains(READY_TO_EXECUTE);
          if (sendUnstable) sendUnstableRead(to, flags);
          else sendStableRead(to, commitKind());
      }
  
      @Override
      protected Ranges unavailable(ReadReply reply)
      {
          return ((ReadOk)reply).unavailable;
      }
  
      @Override
      public void onSuccess(Id from, ReadReply reply)
      {
          super.onSuccess(from, reply);
          // we forward all not-oks for consideration, and any stable fast path read oks;
          // it's up to the stable tracker to decide what to do with them, we're just filtering out unstable Acks
          if (!reply.isOk() || unstableFastReads == null || !unstableFastReads.contains(from))
              stable.onSuccess(from, reply);
      }
  
      @Override
      protected Action process(Id from, ReadReply reply)
      {
          if (reply.isOk())
          {
              ReadOk ok = ((ReadOk) reply);
              Data next = ok.data;
              if (next != null)
                  data = data == null ? next : data.merge(next);
  
              if (txnId.is(Txn.Kind.Write) && ok.uniqueHlc > 0)
              {
                  Invariants.require(ok.uniqueHlc > executeAt.hlc());
                  uniqueHlc = Math.max(uniqueHlc, ok.uniqueHlc);
              }
              return ok.unavailable == null ? Approve : ApprovePartial;
          }
  
          CommitOrReadNack nack = (CommitOrReadNack) reply;
          switch (nack.kind)
          {
              default: throw UnhandledEnum.unknown(nack.kind);
              case InsufficientEpochs: throw UnhandledEnum.invalid(nack.kind);
              case Waiting:
                  if (from.id == node.id().id)
                      isPrivilegedVoteCommitting = false;
                  return Action.None;
  
              case Redundant:
              case Rejected:
                  invokeCallback(null, Preempted.preempted(node.agent(), txnId, route.homeKey()));
                  return Action.Aborted;
  
              case Insufficient:
                  // the replica may be missing the original commit, or the additional commit, so send everything
                  sendMaximal(from);
                  // also try sending a read command to another replica, in case they're ready to serve a response
                  return Action.TryAlternative;
          }
      }
  
      @Override
      protected void onDone(Success success, Throwable failure)
      {
          if (failure == null)
          {
              stable.setDone();
              Timestamp executeAt = this.executeAt;
              if (txnId.is(Txn.Kind.Write) && uniqueHlc != 0)
              {
                  Invariants.require(uniqueHlc > executeAt.hlc());
                  executeAt = new TimestampWithUniqueHlc(executeAt, uniqueHlc);
              }
  
              // Always compute Result before Write to provide integrations with a predictable invocation order
              // in case there is shared state between Result and Update. This can change if really needed
              // just make sure to check the integrations to make sure it won't break anything
              Result result = txn.result(txnId, executeAt, data);
              Writes writes = txnId.is(Txn.Kind.Write) ? txn.execute(txnId, executeAt, data) : null;
              adapter().persist(node, executor, allTopologies, route, ballot, flags, txnId, txn, executeAt, stableDeps, writes, result, takeCallback());
          }
          else
          {
              stable.informStableOnceQuorum();
              if (sendOnlyReadStableMessages())
              {
                  // send additional stable messages to record the transaction outcome
                  Commit.Kind kind = commitKind();
                  if (!candidates.isEmpty())
                  {
                      for (int i = 0, size = candidates.size() ; i < size ; ++i)
                          sendStableOnly(candidates.get(i), kind);
                  }
                  if (unstableFastReads != null)
                  {
                      for (Node.Id to : unstableFastReads)
                          sendStableOnly(to, kind);
                  }
              }
              invokeCallback(null, failure);
          }
      }
  
      @Override
      public void onSlowResponse(Id from)
      {
          // send stable messages to everyone not yet contacted, and then inform decided, to avoid unnecessary recoveries
          stable.maybeInformStable();
          super.onSlowResponse(from);
      }
  
      @Override
      public void onFailure(Id from, Throwable failure)
      {
          super.onFailure(from, failure);
          if (isPrivilegedVoteCommitting && from.id == node.id().id)
              tryFinishOnFailure();
      }
  
      protected CoordinationAdapter<Result> adapter()
      {
          return node.coordinationAdapter(txnId, Standard);
      }
  
      @Override
      public String toString()
      {
          return "ExecuteTxn{" +
                 "txnId=" + txnId +
                 ", txn=" + txn +
                 ", route=" + route +
                 '}';
      }
  
      class LocalExecute extends ReadData
      {
          final boolean mayFastExecute;
          private boolean committed;
          private final SafeCallback<ReadReply> callback;
          private Timeouts.RegisteredTimeout slowTimeout;
  
          public LocalExecute(TxnId txnId, ExecuteFlags flags, boolean mayFastExecute)
          {
              super(txnId, route, mayFastExecute ? txn.intersecting(route, true) : null, ExecuteTxn.this.executeAt, ExecuteTxn.this.executeAt.epoch(), flags);
              this.callback = new SafeCallback<>(executor, ExecuteTxn.this);
              this.mayFastExecute = mayFastExecute;
          }
  
          @Override
          protected boolean mayFastExecute()
          {
              return mayFastExecute;
          }
  
          @Override
          public CommitOrReadNack applyInternal(SafeCommandStore safeStore)
          {
              StoreParticipants participants = StoreParticipants.execute(safeStore, route, txnId, minEpoch(), executeAtEpoch);
              SafeCommand safeCommand = safeStore.get(txnId, participants);
              return applyInternal(safeStore, safeCommand, participants);
          }
  
          @Override
          protected CommitOrReadNack applyInternal(SafeCommandStore safeStore, SafeCommand safeCommand, StoreParticipants participants)
          {
              if (CommitOutcome.Rejected == Commands.commit(safeStore, safeCommand, participants, Stable, Ballot.ZERO, txnId, route, txn, executeAt, stableDeps, commitKind()))
                  return CommitOrReadNack.Rejected;
  
              return super.applyInternal(safeStore, safeCommand, participants);
          }
  
          @Override
          protected CommitOrReadNack refuseInternal(SafeCommandStore safeStore)
          {
              if (isPrivilegedVoteCommitting)
                  throw new LogUnavailableException();
              return super.refuseInternal(safeStore);
          }
  
          @Override
          public void accept(CommitOrReadNack reply, Throwable failure)
          {
              if (reply != null && reply.kind == CommitOrReadNack.Kind.Waiting)
              {
                  committed = true;
                  long slowAt = node.agent().selfSlowAt(txnId, Execute, MICROSECONDS);
                  slowTimeout = node.timeouts().registerAt(new Timeouts.Timeout()
                  {
                      @Override public void timeout() { executor.executeMaybeImmediately(() -> {
                          onSlowResponse(node.id());
                          slowTimeout = null;
                      }); }
                      @Override public int stripe() { return txnId.hashCode(); }
                  }, slowAt, MICROSECONDS);
              }
              else if (failure == null)
              {
                  committed = reply == null && !mayFastExecute;
              }
              super.accept(reply, failure);
          }
  
          @Override
          protected ExecuteOn executeOn()
          {
              return StableThenRead.EXECUTE_ON;
          }
  
          @Override
          protected boolean timeoutInternal()
          {
              if (!super.timeoutInternal())
                  return false;
  
              slowTimeout = null;
              if (committed) callback.failure(node.id(), new Timeout(txnId, route.homeKey(), "Could not promptly read from local coordinator"));
              else callback.failure(node.id(), new Timeout(txnId, route.homeKey(), "Could not promptly commit to local coordinator"));
              return true;
          }
  
          @Override
          protected void reply(ReadReply reply, Throwable fail)
          {
              if (slowTimeout != null && reply != Waiting)
              {
                  slowTimeout.cancel();
                  slowTimeout = null;
              }
  
              // TODO (expected): execute immediately if already on CommandStore
              if (fail == null) callback.success(node.id(), reply);
              else callback.failure(node.id(), fail);
          }
  
          @Override
          public ReadType kind() { throw new UnsupportedOperationException(); }
  
          @Override
          public MessageType type() { throw new UnsupportedOperationException(); }
  
          @Override
          public String reason()
          {
              return "Local Execute";
          }
      }
  
      @Override
      public CoordinationKind kind()
      {
          return CoordinationKind.Execute;
      }
  
      @Override
      public Unseekables<?> scope()
      {
          return route;
      }
  
      @Override
      public String describe()
      {
          // TODO (desired): summarise what data replies we have
          return "path=" + path +
                 ", ballot=" + ballot +
                 ", executeAt=" + executeAt +
                 ", flags=" + flags +
                 ", uniqueHlc=" + uniqueHlc +
                 ", isPrivilegedVoteCommitting=" + isPrivilegedVoteCommitting;
      }
  }
  