docs: private void handleIv2InitiateTaskMessage(Iv2InitiateTaskMessage message)
      {
          if (!message.isSinglePartition()) {
              throw new RuntimeException("SpScheduler.handleIv2InitiateTaskMessage " +
                      "should never receive multi-partition initiations.");
          }
  
          final String procedureName = message.getStoredProcedureName();
          long newSpHandle;
          long uniqueId = Long.MIN_VALUE;
          Iv2InitiateTaskMessage msg = message;
          if (m_isLeader || message.isReadOnly()) {
  
              /*
               * If this is for CL replay or DR, update the unique ID generator
               */
              if (message.isForReplay()) {
                  uniqueId = message.getUniqueId();
                  try {
                      m_uniqueIdGenerator.updateMostRecentlyGeneratedUniqueId(uniqueId);
                  }
                  catch (Exception e) {
                      hostLog.fatal(e.getMessage());
                      hostLog.fatal("Invocation: " + message);
                      VoltDB.crashLocalVoltDB(e.getMessage(), true, e);
                  }
              }
  
              /*
               * If this is CL replay use the txnid from the CL and also
               * update the txnid to match the one from the CL
               */
              if (message.isForReplay()) {
                  TxnEgo ego = advanceTxnEgo();
                  newSpHandle = ego.getTxnId();
                  updateMaxScheduledTransactionSpHandle(newSpHandle);
              } else if (m_isLeader && !message.isReadOnly()) {
                  TxnEgo ego = advanceTxnEgo();
                  newSpHandle = ego.getTxnId();
                  updateMaxScheduledTransactionSpHandle(newSpHandle);
                  uniqueId = m_uniqueIdGenerator.getNextUniqueId();
              } else {
                  /*
                   * The SPI read or the short circuit read case. Since we are read only,
                   * do not create new transaction IDs but reuse the last seen
                   * txnid. For a timestamp, might as well give a reasonable one
                   * for a read heavy workload so time isn't bursty.
                   */
                  uniqueId = UniqueIdGenerator.makeIdFromComponents(
                          Math.max(System.currentTimeMillis(), m_uniqueIdGenerator.lastUsedTime),
                          0,
                          m_uniqueIdGenerator.partitionId);
  
                  newSpHandle = getMaxScheduledTxnSpHandle();
              }
  
              // Need to set the SP handle on the received message
              // Need to copy this or the other local sites handling
              // the same initiate task message will overwrite each
              // other's memory -- the message isn't copied on delivery
              // to other local mailboxes.
              msg = new Iv2InitiateTaskMessage(
                      message.getInitiatorHSId(),
                      message.getCoordinatorHSId(),
                      getRepairLogTruncationHandleForReplicas(),
                      message.getTxnId(),
                      message.getUniqueId(),
                      message.isReadOnly(),
                      message.isSinglePartition(),
                      null,
                      message.getStoredProcedureInvocation(),
                      message.getClientInterfaceHandle(),
                      message.getConnectionId(),
                      message.isForReplay());
  
              msg.setSpHandle(newSpHandle);
              logRepair(msg);
              // Also, if this is a vanilla single-part procedure, make the TXNID
              // be the SpHandle (for now)
              // Only system procedures are every-site, so we'll check through the SystemProcedureCatalog
              if (SystemProcedureCatalog.listing.get(procedureName) == null ||
                      !SystemProcedureCatalog.listing.get(procedureName).getEverysite())
              {
                  msg.setTxnId(newSpHandle);
                  msg.setUniqueId(uniqueId);
              }
  
              // The leader will be responsible to replicate messages to replicas.
              // Don't replicate reads, no matter FAST or SAFE.
              if (m_isLeader && (!msg.isReadOnly()) && IS_KSAFE_CLUSTER ) {
                  for (long hsId : m_sendToHSIds) {
                      Iv2InitiateTaskMessage finalMsg = msg;
                      final VoltTrace.TraceEventBatch traceLog = VoltTrace.log(VoltTrace.Category.SPI);
                      if (traceLog != null) {
                          traceLog.add(() -> VoltTrace.beginAsync("replicateSP",
                                                                  MiscUtils.hsIdPairTxnIdToString(m_mailbox.getHSId(), hsId, finalMsg.getSpHandle(), finalMsg.getClientInterfaceHandle()),
                                                                  "txnId", TxnEgo.txnIdToString(finalMsg.getTxnId()),
                                                                  "dest", CoreUtils.hsIdToString(hsId)));
                      }
                  }
                  Iv2InitiateTaskMessage replmsg =
                      new Iv2InitiateTaskMessage(m_mailbox.getHSId(),
                              m_mailbox.getHSId(),
                              getRepairLogTruncationHandleForReplicas(),
                              msg.getTxnId(),
                              msg.getUniqueId(),
                              msg.isReadOnly(),
                              msg.isSinglePartition(),
                              msg.getStoredProcedureInvocation(),
                              msg.getClientInterfaceHandle(),
                              msg.getConnectionId(),
                              msg.isForReplay(),
                              true);
                  // Update the handle in the copy since the constructor doesn't set it
                  replmsg.setSpHandle(newSpHandle);
                  // K-safety cluster doesn't always mean partition has replicas,
                  // node failure may reduce the number of replicas for each partition
                  if (m_sendToHSIds.length > 0) {
                      m_mailbox.send(m_sendToHSIds, replmsg);
                  }
  
                  DuplicateCounter counter = new DuplicateCounter(
                          msg.getInitiatorHSId(),
                          msg.getTxnId(),
                          m_replicaHSIds,
                          replmsg);
  
                  safeAddToDuplicateCounterMap(new DuplicateCounterKey(msg.getTxnId(), newSpHandle), counter);
              }
          }
          else {
              setMaxSeenTxnId(msg.getSpHandle());
              newSpHandle = msg.getSpHandle();
              logRepair(msg);
              // Don't update the uniqueID if this is a run-everywhere txn, because it has an MPI unique ID.
              if (UniqueIdGenerator.getPartitionIdFromUniqueId(msg.getUniqueId()) == m_partitionId) {
                  m_uniqueIdGenerator.updateMostRecentlyGeneratedUniqueId(msg.getUniqueId());
              }
          }
          Iv2Trace.logIv2InitiateTaskMessage(message, m_mailbox.getHSId(), msg.getTxnId(), newSpHandle);
          doLocalInitiateOffer(msg);
      }