<commit_before>/*
Copyright 2019 The Vitess Authors.

Licensed 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 tabletmanager

import (
	"flag"
	"fmt"
	"strconv"
	"strings"
	"time"

	"vitess.io/vitess/go/vt/proto/vtrpc"

	"context"

	"vitess.io/vitess/go/mysql"
	"vitess.io/vitess/go/vt/log"
	"vitess.io/vitess/go/vt/logutil"
	"vitess.io/vitess/go/vt/mysqlctl"
	"vitess.io/vitess/go/vt/topo/topoproto"
	"vitess.io/vitess/go/vt/vterrors"

	replicationdatapb "vitess.io/vitess/go/vt/proto/replicationdata"
	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
)

var (
	enableSemiSync   = flag.Bool("enable_semi_sync", false, "Enable semi-sync when configuring replication, on primary and replica tablets only (rdonly tablets will not ack).")
	setSuperReadOnly = flag.Bool("use_super_read_only", true, "Set super_read_only flag when performing planned failover.")
)

// ReplicationStatus returns the replication status
func (tm *TabletManager) ReplicationStatus(ctx context.Context) (*replicationdatapb.Status, error) {
	status, err := tm.MysqlDaemon.ReplicationStatus()
	if err != nil {
		return nil, err
	}
	return mysql.ReplicationStatusToProto(status), nil
}

// MasterStatus is the old version of PrimaryStatus. Deprecated.
func (tm *TabletManager) MasterStatus(ctx context.Context) (*replicationdatapb.PrimaryStatus, error) {
	return tm.PrimaryStatus(ctx)
}

// PrimaryStatus returns the replication status for a primary tablet.
func (tm *TabletManager) PrimaryStatus(ctx context.Context) (*replicationdatapb.PrimaryStatus, error) {
	status, err := tm.MysqlDaemon.PrimaryStatus(ctx)
	if err != nil {
		return nil, err
	}
	return mysql.PrimaryStatusToProto(status), nil
}

// MasterPosition is the old version of PrimaryPosition. Deprecated.
func (tm *TabletManager) MasterPosition(ctx context.Context) (string, error) {
	return tm.PrimaryPosition(ctx)
}

// PrimaryPosition returns the position of a primary database
func (tm *TabletManager) PrimaryPosition(ctx context.Context) (string, error) {
	pos, err := tm.MysqlDaemon.PrimaryPosition()
	if err != nil {
		return "", err
	}
	return mysql.EncodePosition(pos), nil
}

// WaitForPosition waits until replication reaches the desired position
func (tm *TabletManager) WaitForPosition(ctx context.Context, pos string) error {
	log.Infof("WaitForPosition: %v", pos)
	mpos, err := mysql.DecodePosition(pos)
	if err != nil {
		return err
	}
	return tm.MysqlDaemon.WaitSourcePos(ctx, mpos)
}

// StopReplication will stop the mysql. Works both when Vitess manages
// replication or not (using hook if not).
func (tm *TabletManager) StopReplication(ctx context.Context) error {
	log.Infof("StopReplication")
	if err := tm.lock(ctx); err != nil {
		return err
	}
	defer tm.unlock()

	return tm.stopReplicationLocked(ctx)
}

func (tm *TabletManager) stopReplicationLocked(ctx context.Context) error {

	// Remember that we were told to stop, so we don't try to
	// restart ourselves (in replication_reporter).
	tm.replManager.setReplicationStopped(true)

	// Also tell Orchestrator we're stopped on purpose for some Vitess task.
	// Do this in the background, as it's best-effort.
	go func() {
		if tm.orc == nil {
			return
		}
		if err := tm.orc.BeginMaintenance(tm.Tablet(), "vttablet has been told to StopReplication"); err != nil {
			log.Warningf("Orchestrator BeginMaintenance failed: %v", err)
		}
	}()

	return tm.MysqlDaemon.StopReplication(tm.hookExtraEnv())
}

func (tm *TabletManager) stopIOThreadLocked(ctx context.Context) error {

	// Remember that we were told to stop, so we don't try to
	// restart ourselves (in replication_reporter).
	tm.replManager.setReplicationStopped(true)

	// Also tell Orchestrator we're stopped on purpose for some Vitess task.
	// Do this in the background, as it's best-effort.
	go func() {
		if tm.orc == nil {
			return
		}
		if err := tm.orc.BeginMaintenance(tm.Tablet(), "vttablet has been told to StopReplication"); err != nil {
			log.Warningf("Orchestrator BeginMaintenance failed: %v", err)
		}
	}()

	return tm.MysqlDaemon.StopIOThread(ctx)
}

// StopReplicationMinimum will stop the replication after it reaches at least the
// provided position. Works both when Vitess manages
// replication or not (using hook if not).
func (tm *TabletManager) StopReplicationMinimum(ctx context.Context, position string, waitTime time.Duration) (string, error) {
	log.Infof("StopReplicationMinimum: position: %v waitTime: %v", position, waitTime)
	if err := tm.lock(ctx); err != nil {
		return "", err
	}
	defer tm.unlock()

	pos, err := mysql.DecodePosition(position)
	if err != nil {
		return "", err
	}
	waitCtx, cancel := context.WithTimeout(ctx, waitTime)
	defer cancel()
	if err := tm.MysqlDaemon.WaitSourcePos(waitCtx, pos); err != nil {
		return "", err
	}
	if err := tm.stopReplicationLocked(ctx); err != nil {
		return "", err
	}
	pos, err = tm.MysqlDaemon.PrimaryPosition()
	if err != nil {
		return "", err
	}
	return mysql.EncodePosition(pos), nil
}

// StartReplication will start the mysql. Works both when Vitess manages
// replication or not (using hook if not).
func (tm *TabletManager) StartReplication(ctx context.Context, semiSync bool) error {
	log.Infof("StartReplication")
	if err := tm.lock(ctx); err != nil {
		return err
	}
	defer tm.unlock()

	tm.replManager.setReplicationStopped(false)

	// Tell Orchestrator we're no longer stopped on purpose.
	// Do this in the background, as it's best-effort.
	go func() {
		if tm.orc == nil {
			return
		}
		if err := tm.orc.EndMaintenance(tm.Tablet()); err != nil {
			log.Warningf("Orchestrator EndMaintenance failed: %v", err)
		}
	}()

	if err := tm.fixSemiSync(tm.Tablet().Type, convertBoolToSemiSyncAction(semiSync)); err != nil {
		return err
	}
	return tm.MysqlDaemon.StartReplication(tm.hookExtraEnv())
}

// StartReplicationUntilAfter will start the replication and let it catch up
// until and including the transactions in `position`
func (tm *TabletManager) StartReplicationUntilAfter(ctx context.Context, position string, waitTime time.Duration) error {
	log.Infof("StartReplicationUntilAfter: position: %v waitTime: %v", position, waitTime)
	if err := tm.lock(ctx); err != nil {
		return err
	}
	defer tm.unlock()

	waitCtx, cancel := context.WithTimeout(ctx, waitTime)
	defer cancel()

	pos, err := mysql.DecodePosition(position)
	if err != nil {
		return err
	}

	return tm.MysqlDaemon.StartReplicationUntilAfter(waitCtx, pos)
}

// GetReplicas returns the address of all the replicas
func (tm *TabletManager) GetReplicas(ctx context.Context) ([]string, error) {
	return mysqlctl.FindReplicas(tm.MysqlDaemon)
}

// ResetReplication completely resets the replication on the host.
// All binary and relay logs are flushed. All replication positions are reset.
func (tm *TabletManager) ResetReplication(ctx context.Context) error {
	log.Infof("ResetReplication")
	if err := tm.lock(ctx); err != nil {
		return err
	}
	defer tm.unlock()

	tm.replManager.setReplicationStopped(true)
	return tm.MysqlDaemon.ResetReplication(ctx)
}

// InitMaster is the old version of InitPrimary. Deprecated.
func (tm *TabletManager) InitMaster(ctx context.Context, semiSync bool) (string, error) {
	return tm.InitPrimary(ctx, semiSync)
}

// InitPrimary enables writes and returns the replication position.
func (tm *TabletManager) InitPrimary(ctx context.Context, semiSync bool) (string, error) {
	log.Infof("InitPrimary")
	if err := tm.lock(ctx); err != nil {
		return "", err
	}
	defer tm.unlock()

	// Initializing as primary implies undoing any previous "do not replicate".
	tm.replManager.reset()

	if *setSuperReadOnly {
		// Setting super_read_only off so that we can run the DDL commands
		if err := tm.MysqlDaemon.SetSuperReadOnly(false); err != nil {
			if strings.Contains(err.Error(), strconv.Itoa(mysql.ERUnknownSystemVariable)) {
				log.Warningf("server does not know about super_read_only, continuing anyway...")
			} else {
				return "", err
			}
		}
	}

	// we need to insert something in the binlogs, so we can get the
	// current position. Let's just use the mysqlctl.CreateReparentJournal commands.
	cmds := mysqlctl.CreateReparentJournal()
	if err := tm.MysqlDaemon.ExecuteSuperQueryList(ctx, cmds); err != nil {
		return "", err
	}

	// Execute ALTER statement on reparent_journal table and ignore errors
	cmds = mysqlctl.AlterReparentJournal()
	_ = tm.MysqlDaemon.ExecuteSuperQueryList(ctx, cmds)

	// get the current replication position
	pos, err := tm.MysqlDaemon.PrimaryPosition()
	if err != nil {
		return "", err
	}

	// Set the server read-write, from now on we can accept real
	// client writes. Note that if semi-sync replication is enabled,
	// we'll still need some replicas to be able to commit transactions.
	if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_PRIMARY, DBActionSetReadWrite, convertBoolToSemiSyncAction(semiSync)); err != nil {
		return "", err
	}

	// Enforce semi-sync after changing the tablet)type to PRIMARY. Otherwise, the
	// primary will hang while trying to create the database.
	if err := tm.fixSemiSync(topodatapb.TabletType_PRIMARY, convertBoolToSemiSyncAction(semiSync)); err != nil {
		return "", err
	}

	return mysql.EncodePosition(pos), nil
}

// PopulateReparentJournal adds an entry into the reparent_journal table.
func (tm *TabletManager) PopulateReparentJournal(ctx context.Context, timeCreatedNS int64, actionName string, primaryAlias *topodatapb.TabletAlias, position string) error {
	log.Infof("PopulateReparentJournal: action: %v parent: %v  position: %v", actionName, primaryAlias, position)
	pos, err := mysql.DecodePosition(position)
	if err != nil {
		return err
	}
	cmds := mysqlctl.CreateReparentJournal()
	if err := tm.MysqlDaemon.ExecuteSuperQueryList(ctx, cmds); err != nil {
		return err
	}

	// Execute ALTER statement on reparent_journal table and ignore errors
	cmds = mysqlctl.AlterReparentJournal()
	_ = tm.MysqlDaemon.ExecuteSuperQueryList(ctx, cmds)

	cmds = []string{mysqlctl.PopulateReparentJournal(timeCreatedNS, actionName, topoproto.TabletAliasString(primaryAlias), pos)}

	return tm.MysqlDaemon.ExecuteSuperQueryList(ctx, cmds)
}

// InitReplica sets replication primary and position, and waits for the
// reparent_journal table entry up to context timeout
func (tm *TabletManager) InitReplica(ctx context.Context, parent *topodatapb.TabletAlias, position string, timeCreatedNS int64, semiSync bool) error {
	log.Infof("InitReplica: parent: %v  position: %v", parent, position)
	if err := tm.lock(ctx); err != nil {
		return err
	}
	defer tm.unlock()

	// If we were a primary type, switch our type to replica.  This
	// is used on the old primary when using InitShardPrimary with
	// -force, and the new primary is different from the old primary.
	if tm.Tablet().Type == topodatapb.TabletType_PRIMARY {
		if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_REPLICA, DBActionNone, convertBoolToSemiSyncAction(semiSync)); err != nil {
			return err
		}
	}

	pos, err := mysql.DecodePosition(position)
	if err != nil {
		return err
	}
	ti, err := tm.TopoServer.GetTablet(ctx, parent)
	if err != nil {
		return err
	}

	tm.replManager.setReplicationStopped(false)

	// If using semi-sync, we need to enable it before connecting to primary.
	// If we were a primary type, we need to switch back to replica settings.
	// Otherwise we won't be able to commit anything.
	tt := tm.Tablet().Type
	if tt == topodatapb.TabletType_PRIMARY {
		tt = topodatapb.TabletType_REPLICA
	}
	if err := tm.fixSemiSync(tt, convertBoolToSemiSyncAction(semiSync)); err != nil {
		return err
	}

	if err := tm.MysqlDaemon.SetReplicationPosition(ctx, pos); err != nil {
		return err
	}
	if err := tm.MysqlDaemon.SetReplicationSource(ctx, ti.Tablet.MysqlHostname, int(ti.Tablet.MysqlPort), false /* stopReplicationBefore */, true /* stopReplicationAfter */); err != nil {
		return err
	}

	// wait until we get the replicated row, or our context times out
	return tm.MysqlDaemon.WaitForReparentJournal(ctx, timeCreatedNS)
}

// DemotePrimary prepares a PRIMARY tablet to give up leadership to another tablet.
//
// It attemps to idempotently ensure the following guarantees upon returning
// successfully:
//   * No future writes will be accepted.
//   * No writes are in-flight.
//   * MySQL is in read-only mode.
//   * Semi-sync settings are consistent with a REPLICA tablet.
//
// If necessary, it waits for all in-flight writes to complete or time out.
//
// It should be safe to call this on a PRIMARY tablet that was already demoted,
// or on a tablet that already transitioned to REPLICA.
//
// If a step fails in the middle, it will try to undo any changes it made.
func (tm *TabletManager) DemotePrimary(ctx context.Context) (*replicationdatapb.PrimaryStatus, error) {
	log.Infof("DemotePrimary")
	// The public version always reverts on partial failure.
	return tm.demotePrimary(ctx, true /* revertPartialFailure */)
}

// DemoteMaster is the old version of DemotePrimary. Deprecated.
func (tm *TabletManager) DemoteMaster(ctx context.Context) (*replicationdatapb.PrimaryStatus, error) {
	return tm.DemotePrimary(ctx)
}

// demotePrimary implements DemotePrimary with an additional, private option.
//
// If revertPartialFailure is true, and a step fails in the middle, it will try
// to undo any changes it made.
func (tm *TabletManager) demotePrimary(ctx context.Context, revertPartialFailure bool) (primaryStatus *replicationdatapb.PrimaryStatus, finalErr error) {
	if err := tm.lock(ctx); err != nil {
		return nil, err
	}
	defer tm.unlock()

	tablet := tm.Tablet()
	wasPrimary := tablet.Type == topodatapb.TabletType_PRIMARY
	wasServing := tm.QueryServiceControl.IsServing()
	wasReadOnly, err := tm.MysqlDaemon.IsReadOnly()
	if err != nil {
		return nil, err
	}

	// If we are a primary tablet and not yet read-only, stop accepting new
	// queries and wait for in-flight queries to complete. If we are not primary,
	// or if we are already read-only, there's no need to stop the queryservice
	// in order to ensure the guarantee we are being asked to provide, which is
	// that no writes are occurring.
	if wasPrimary && !wasReadOnly {
		// Tell Orchestrator we're stopped on purpose for demotion.
		// This is a best effort task, so run it in a goroutine.
		go func() {
			if tm.orc == nil {
				return
			}
			if err := tm.orc.BeginMaintenance(tm.Tablet(), "vttablet has been told to DemotePrimary"); err != nil {
				log.Warningf("Orchestrator BeginMaintenance failed: %v", err)
			}
		}()

		// Note that this may block until the transaction timeout if clients
		// don't finish their transactions in time. Even if some transactions
		// have to be killed at the end of their timeout, this will be
		// considered successful. If we are already not serving, this will be
		// idempotent.
		log.Infof("DemotePrimary disabling query service")
		if err := tm.QueryServiceControl.SetServingType(tablet.Type, logutil.ProtoToTime(tablet.PrimaryTermStartTime), false, "demotion in progress"); err != nil {
			return nil, vterrors.Wrap(err, "SetServingType(serving=false) failed")
		}
		defer func() {
			if finalErr != nil && revertPartialFailure && wasServing {
				if err := tm.QueryServiceControl.SetServingType(tablet.Type, logutil.ProtoToTime(tablet.PrimaryTermStartTime), true, ""); err != nil {
					log.Warningf("SetServingType(serving=true) failed during revert: %v", err)
				}
			}
		}()
	}

	// Now that we know no writes are in-flight and no new writes can occur,
	// set MySQL to read-only mode. If we are already read-only because of a
	// previous demotion, or because we are not primary anyway, this should be
	// idempotent.
	if *setSuperReadOnly {
		// Setting super_read_only also sets read_only
		if err := tm.MysqlDaemon.SetSuperReadOnly(true); err != nil {
			if strings.Contains(err.Error(), strconv.Itoa(mysql.ERUnknownSystemVariable)) {
				log.Warningf("server does not know about super_read_only, continuing anyway...")
			} else {
				return nil, err
			}
		}
	} else {
		if err := tm.MysqlDaemon.SetReadOnly(true); err != nil {
			return nil, err
		}
	}
	defer func() {
		if finalErr != nil && revertPartialFailure && !wasReadOnly {
			// setting read_only OFF will also set super_read_only OFF if it was set
			if err := tm.MysqlDaemon.SetReadOnly(false); err != nil {
				log.Warningf("SetReadOnly(false) failed during revert: %v", err)
			}
		}
	}()

	if tm.isPrimarySideSemiSyncEnabled() {
		// If using semi-sync, we need to disable primary-side.
		if err := tm.fixSemiSync(topodatapb.TabletType_REPLICA, SemiSyncActionSet); err != nil {
			return nil, err
		}
		defer func() {
			if finalErr != nil && revertPartialFailure && wasPrimary {
				// enable primary-side semi-sync again<commit_msg>refactor: add comment explaining the use of isPrimarySideSemiSyncEnabled<commit_after>/*
Copyright 2019 The Vitess Authors.

Licensed 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 tabletmanager

import (
	"flag"
	"fmt"
	"strconv"
	"strings"
	"time"

	"vitess.io/vitess/go/vt/proto/vtrpc"

	"context"

	"vitess.io/vitess/go/mysql"
	"vitess.io/vitess/go/vt/log"
	"vitess.io/vitess/go/vt/logutil"
	"vitess.io/vitess/go/vt/mysqlctl"
	"vitess.io/vitess/go/vt/topo/topoproto"
	"vitess.io/vitess/go/vt/vterrors"

	replicationdatapb "vitess.io/vitess/go/vt/proto/replicationdata"
	topodatapb "vitess.io/vitess/go/vt/proto/topodata"
)

var (
	enableSemiSync   = flag.Bool("enable_semi_sync", false, "Enable semi-sync when configuring replication, on primary and replica tablets only (rdonly tablets will not ack).")
	setSuperReadOnly = flag.Bool("use_super_read_only", true, "Set super_read_only flag when performing planned failover.")
)

// ReplicationStatus returns the replication status
func (tm *TabletManager) ReplicationStatus(ctx context.Context) (*replicationdatapb.Status, error) {
	status, err := tm.MysqlDaemon.ReplicationStatus()
	if err != nil {
		return nil, err
	}
	return mysql.ReplicationStatusToProto(status), nil
}

// MasterStatus is the old version of PrimaryStatus. Deprecated.
func (tm *TabletManager) MasterStatus(ctx context.Context) (*replicationdatapb.PrimaryStatus, error) {
	return tm.PrimaryStatus(ctx)
}

// PrimaryStatus returns the replication status for a primary tablet.
func (tm *TabletManager) PrimaryStatus(ctx context.Context) (*replicationdatapb.PrimaryStatus, error) {
	status, err := tm.MysqlDaemon.PrimaryStatus(ctx)
	if err != nil {
		return nil, err
	}
	return mysql.PrimaryStatusToProto(status), nil
}

// MasterPosition is the old version of PrimaryPosition. Deprecated.
func (tm *TabletManager) MasterPosition(ctx context.Context) (string, error) {
	return tm.PrimaryPosition(ctx)
}

// PrimaryPosition returns the position of a primary database
func (tm *TabletManager) PrimaryPosition(ctx context.Context) (string, error) {
	pos, err := tm.MysqlDaemon.PrimaryPosition()
	if err != nil {
		return "", err
	}
	return mysql.EncodePosition(pos), nil
}

// WaitForPosition waits until replication reaches the desired position
func (tm *TabletManager) WaitForPosition(ctx context.Context, pos string) error {
	log.Infof("WaitForPosition: %v", pos)
	mpos, err := mysql.DecodePosition(pos)
	if err != nil {
		return err
	}
	return tm.MysqlDaemon.WaitSourcePos(ctx, mpos)
}

// StopReplication will stop the mysql. Works both when Vitess manages
// replication or not (using hook if not).
func (tm *TabletManager) StopReplication(ctx context.Context) error {
	log.Infof("StopReplication")
	if err := tm.lock(ctx); err != nil {
		return err
	}
	defer tm.unlock()

	return tm.stopReplicationLocked(ctx)
}

func (tm *TabletManager) stopReplicationLocked(ctx context.Context) error {

	// Remember that we were told to stop, so we don't try to
	// restart ourselves (in replication_reporter).
	tm.replManager.setReplicationStopped(true)

	// Also tell Orchestrator we're stopped on purpose for some Vitess task.
	// Do this in the background, as it's best-effort.
	go func() {
		if tm.orc == nil {
			return
		}
		if err := tm.orc.BeginMaintenance(tm.Tablet(), "vttablet has been told to StopReplication"); err != nil {
			log.Warningf("Orchestrator BeginMaintenance failed: %v", err)
		}
	}()

	return tm.MysqlDaemon.StopReplication(tm.hookExtraEnv())
}

func (tm *TabletManager) stopIOThreadLocked(ctx context.Context) error {

	// Remember that we were told to stop, so we don't try to
	// restart ourselves (in replication_reporter).
	tm.replManager.setReplicationStopped(true)

	// Also tell Orchestrator we're stopped on purpose for some Vitess task.
	// Do this in the background, as it's best-effort.
	go func() {
		if tm.orc == nil {
			return
		}
		if err := tm.orc.BeginMaintenance(tm.Tablet(), "vttablet has been told to StopReplication"); err != nil {
			log.Warningf("Orchestrator BeginMaintenance failed: %v", err)
		}
	}()

	return tm.MysqlDaemon.StopIOThread(ctx)
}

// StopReplicationMinimum will stop the replication after it reaches at least the
// provided position. Works both when Vitess manages
// replication or not (using hook if not).
func (tm *TabletManager) StopReplicationMinimum(ctx context.Context, position string, waitTime time.Duration) (string, error) {
	log.Infof("StopReplicationMinimum: position: %v waitTime: %v", position, waitTime)
	if err := tm.lock(ctx); err != nil {
		return "", err
	}
	defer tm.unlock()

	pos, err := mysql.DecodePosition(position)
	if err != nil {
		return "", err
	}
	waitCtx, cancel := context.WithTimeout(ctx, waitTime)
	defer cancel()
	if err := tm.MysqlDaemon.WaitSourcePos(waitCtx, pos); err != nil {
		return "", err
	}
	if err := tm.stopReplicationLocked(ctx); err != nil {
		return "", err
	}
	pos, err = tm.MysqlDaemon.PrimaryPosition()
	if err != nil {
		return "", err
	}
	return mysql.EncodePosition(pos), nil
}

// StartReplication will start the mysql. Works both when Vitess manages
// replication or not (using hook if not).
func (tm *TabletManager) StartReplication(ctx context.Context, semiSync bool) error {
	log.Infof("StartReplication")
	if err := tm.lock(ctx); err != nil {
		return err
	}
	defer tm.unlock()

	tm.replManager.setReplicationStopped(false)

	// Tell Orchestrator we're no longer stopped on purpose.
	// Do this in the background, as it's best-effort.
	go func() {
		if tm.orc == nil {
			return
		}
		if err := tm.orc.EndMaintenance(tm.Tablet()); err != nil {
			log.Warningf("Orchestrator EndMaintenance failed: %v", err)
		}
	}()

	if err := tm.fixSemiSync(tm.Tablet().Type, convertBoolToSemiSyncAction(semiSync)); err != nil {
		return err
	}
	return tm.MysqlDaemon.StartReplication(tm.hookExtraEnv())
}

// StartReplicationUntilAfter will start the replication and let it catch up
// until and including the transactions in `position`
func (tm *TabletManager) StartReplicationUntilAfter(ctx context.Context, position string, waitTime time.Duration) error {
	log.Infof("StartReplicationUntilAfter: position: %v waitTime: %v", position, waitTime)
	if err := tm.lock(ctx); err != nil {
		return err
	}
	defer tm.unlock()

	waitCtx, cancel := context.WithTimeout(ctx, waitTime)
	defer cancel()

	pos, err := mysql.DecodePosition(position)
	if err != nil {
		return err
	}

	return tm.MysqlDaemon.StartReplicationUntilAfter(waitCtx, pos)
}

// GetReplicas returns the address of all the replicas
func (tm *TabletManager) GetReplicas(ctx context.Context) ([]string, error) {
	return mysqlctl.FindReplicas(tm.MysqlDaemon)
}

// ResetReplication completely resets the replication on the host.
// All binary and relay logs are flushed. All replication positions are reset.
func (tm *TabletManager) ResetReplication(ctx context.Context) error {
	log.Infof("ResetReplication")
	if err := tm.lock(ctx); err != nil {
		return err
	}
	defer tm.unlock()

	tm.replManager.setReplicationStopped(true)
	return tm.MysqlDaemon.ResetReplication(ctx)
}

// InitMaster is the old version of InitPrimary. Deprecated.
func (tm *TabletManager) InitMaster(ctx context.Context, semiSync bool) (string, error) {
	return tm.InitPrimary(ctx, semiSync)
}

// InitPrimary enables writes and returns the replication position.
func (tm *TabletManager) InitPrimary(ctx context.Context, semiSync bool) (string, error) {
	log.Infof("InitPrimary")
	if err := tm.lock(ctx); err != nil {
		return "", err
	}
	defer tm.unlock()

	// Initializing as primary implies undoing any previous "do not replicate".
	tm.replManager.reset()

	if *setSuperReadOnly {
		// Setting super_read_only off so that we can run the DDL commands
		if err := tm.MysqlDaemon.SetSuperReadOnly(false); err != nil {
			if strings.Contains(err.Error(), strconv.Itoa(mysql.ERUnknownSystemVariable)) {
				log.Warningf("server does not know about super_read_only, continuing anyway...")
			} else {
				return "", err
			}
		}
	}

	// we need to insert something in the binlogs, so we can get the
	// current position. Let's just use the mysqlctl.CreateReparentJournal commands.
	cmds := mysqlctl.CreateReparentJournal()
	if err := tm.MysqlDaemon.ExecuteSuperQueryList(ctx, cmds); err != nil {
		return "", err
	}

	// Execute ALTER statement on reparent_journal table and ignore errors
	cmds = mysqlctl.AlterReparentJournal()
	_ = tm.MysqlDaemon.ExecuteSuperQueryList(ctx, cmds)

	// get the current replication position
	pos, err := tm.MysqlDaemon.PrimaryPosition()
	if err != nil {
		return "", err
	}

	// Set the server read-write, from now on we can accept real
	// client writes. Note that if semi-sync replication is enabled,
	// we'll still need some replicas to be able to commit transactions.
	if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_PRIMARY, DBActionSetReadWrite, convertBoolToSemiSyncAction(semiSync)); err != nil {
		return "", err
	}

	// Enforce semi-sync after changing the tablet)type to PRIMARY. Otherwise, the
	// primary will hang while trying to create the database.
	if err := tm.fixSemiSync(topodatapb.TabletType_PRIMARY, convertBoolToSemiSyncAction(semiSync)); err != nil {
		return "", err
	}

	return mysql.EncodePosition(pos), nil
}

// PopulateReparentJournal adds an entry into the reparent_journal table.
func (tm *TabletManager) PopulateReparentJournal(ctx context.Context, timeCreatedNS int64, actionName string, primaryAlias *topodatapb.TabletAlias, position string) error {
	log.Infof("PopulateReparentJournal: action: %v parent: %v  position: %v", actionName, primaryAlias, position)
	pos, err := mysql.DecodePosition(position)
	if err != nil {
		return err
	}
	cmds := mysqlctl.CreateReparentJournal()
	if err := tm.MysqlDaemon.ExecuteSuperQueryList(ctx, cmds); err != nil {
		return err
	}

	// Execute ALTER statement on reparent_journal table and ignore errors
	cmds = mysqlctl.AlterReparentJournal()
	_ = tm.MysqlDaemon.ExecuteSuperQueryList(ctx, cmds)

	cmds = []string{mysqlctl.PopulateReparentJournal(timeCreatedNS, actionName, topoproto.TabletAliasString(primaryAlias), pos)}

	return tm.MysqlDaemon.ExecuteSuperQueryList(ctx, cmds)
}

// InitReplica sets replication primary and position, and waits for the
// reparent_journal table entry up to context timeout
func (tm *TabletManager) InitReplica(ctx context.Context, parent *topodatapb.TabletAlias, position string, timeCreatedNS int64, semiSync bool) error {
	log.Infof("InitReplica: parent: %v  position: %v", parent, position)
	if err := tm.lock(ctx); err != nil {
		return err
	}
	defer tm.unlock()

	// If we were a primary type, switch our type to replica.  This
	// is used on the old primary when using InitShardPrimary with
	// -force, and the new primary is different from the old primary.
	if tm.Tablet().Type == topodatapb.TabletType_PRIMARY {
		if err := tm.changeTypeLocked(ctx, topodatapb.TabletType_REPLICA, DBActionNone, convertBoolToSemiSyncAction(semiSync)); err != nil {
			return err
		}
	}

	pos, err := mysql.DecodePosition(position)
	if err != nil {
		return err
	}
	ti, err := tm.TopoServer.GetTablet(ctx, parent)
	if err != nil {
		return err
	}

	tm.replManager.setReplicationStopped(false)

	// If using semi-sync, we need to enable it before connecting to primary.
	// If we were a primary type, we need to switch back to replica settings.
	// Otherwise we won't be able to commit anything.
	tt := tm.Tablet().Type
	if tt == topodatapb.TabletType_PRIMARY {
		tt = topodatapb.TabletType_REPLICA
	}
	if err := tm.fixSemiSync(tt, convertBoolToSemiSyncAction(semiSync)); err != nil {
		return err
	}

	if err := tm.MysqlDaemon.SetReplicationPosition(ctx, pos); err != nil {
		return err
	}
	if err := tm.MysqlDaemon.SetReplicationSource(ctx, ti.Tablet.MysqlHostname, int(ti.Tablet.MysqlPort), false /* stopReplicationBefore */, true /* stopReplicationAfter */); err != nil {
		return err
	}

	// wait until we get the replicated row, or our context times out
	return tm.MysqlDaemon.WaitForReparentJournal(ctx, timeCreatedNS)
}

// DemotePrimary prepares a PRIMARY tablet to give up leadership to another tablet.
//
// It attemps to idempotently ensure the following guarantees upon returning
// successfully:
//   * No future writes will be accepted.
//   * No writes are in-flight.
//   * MySQL is in read-only mode.
//   * Semi-sync settings are consistent with a REPLICA tablet.
//
// If necessary, it waits for all in-flight writes to complete or time out.
//
// It should be safe to call this on a PRIMARY tablet that was already demoted,
// or on a tablet that already transitioned to REPLICA.
//
// If a step fails in the middle, it will try to undo any changes it made.
func (tm *TabletManager) DemotePrimary(ctx context.Context) (*replicationdatapb.PrimaryStatus, error) {
	log.Infof("DemotePrimary")
	// The public version always reverts on partial failure.
	return tm.demotePrimary(ctx, true /* revertPartialFailure */)
}

// DemoteMaster is the old version of DemotePrimary. Deprecated.
func (tm *TabletManager) DemoteMaster(ctx context.Context) (*replicationdatapb.PrimaryStatus, error) {
	return tm.DemotePrimary(ctx)
}

// demotePrimary implements DemotePrimary with an additional, private option.
//
// If revertPartialFailure is true, and a step fails in the middle, it will try
// to undo any changes it made.
func (tm *TabletManager) demotePrimary(ctx context.Context, revertPartialFailure bool) (primaryStatus *replicationdatapb.PrimaryStatus, finalErr error) {
	if err := tm.lock(ctx); err != nil {
		return nil, err
	}
	defer tm.unlock()

	tablet := tm.Tablet()
	wasPrimary := tablet.Type == topodatapb.TabletType_PRIMARY
	wasServing := tm.QueryServiceControl.IsServing()
	wasReadOnly, err := tm.MysqlDaemon.IsReadOnly()
	if err != nil {
		return nil, err
	}

	// If we are a primary tablet and not yet read-only, stop accepting new
	// queries and wait for in-flight queries to complete. If we are not primary,
	// or if we are already read-only, there's no need to stop the queryservice
	// in order to ensure the guarantee we are being asked to provide, which is
	// that no writes are occurring.
	if wasPrimary && !wasReadOnly {
		// Tell Orchestrator we're stopped on purpose for demotion.
		// This is a best effort task, so run it in a goroutine.
		go func() {
			if tm.orc == nil {
				return
			}
			if err := tm.orc.BeginMaintenance(tm.Tablet(), "vttablet has been told to DemotePrimary"); err != nil {
				log.Warningf("Orchestrator BeginMaintenance failed: %v", err)
			}
		}()

		// Note that this may block until the transaction timeout if clients
		// don't finish their transactions in time. Even if some transactions
		// have to be killed at the end of their timeout, this will be
		// considered successful. If we are already not serving, this will be
		// idempotent.
		log.Infof("DemotePrimary disabling query service")
		if err := tm.QueryServiceControl.SetServingType(tablet.Type, logutil.ProtoToTime(tablet.PrimaryTermStartTime), false, "demotion in progress"); err != nil {
			return nil, vterrors.Wrap(err, "SetServingType(serving=false) failed")
		}
		defer func() {
			if finalErr != nil && revertPartialFailure && wasServing {
				if err := tm.QueryServiceControl.SetServingType(tablet.Type, logutil.ProtoToTime(tablet.PrimaryTermStartTime), true, ""); err != nil {
					log.Warningf("SetServingType(serving=true) failed during revert: %v", err)
				}
			}
		}()
	}

	// Now that we know no writes are in-flight and no new writes can occur,
	// set MySQL to read-only mode. If we are already read-only because of a
	// previous demotion, or because we are not primary anyway, this should be
	// idempotent.
	if *setSuperReadOnly {
		// Setting super_read_only also sets read_only
		if err := tm.MysqlDaemon.SetSuperReadOnly(true); err != nil {
			if strings.Contains(err.Error(), strconv.Itoa(mysql.ERUnknownSystemVariable)) {
				log.Warningf("server does not know about super_read_only, continuing anyway...")
			} else {
				return nil, err
			}
		}
	} else {
		if err := tm.MysqlDaemon.SetReadOnly(true); err != nil {
			return nil, err
		}
	}
	defer func() {
		if finalErr != nil && revertPartialFailure && !wasReadOnly {
			// setting read_only OFF will also set super_read_only OFF if it was set
			if err := tm.MysqlDaemon.SetReadOnly(false); err != nil {
				log.Warningf("SetReadOnly(false) failed during revert: %v", err)
			}
		}
	}()

	// Here, we check if the primary side semi sync is enabled or not. If it isn't enabled then we do not need to take any action.
	// If it is enabled then we should turn it off and revert in case of failure.
	if tm.isPrimarySideSemiSyncEnabled() {
		// If using semi-sync, we need to disable primary-side.
		if err := tm.fixSemiSync(topodatapb.TabletType_REPLICA, SemiSyncActionSet); err != nil {
			return nil, err
		}
		defer func() {
			if finalErr != nil && revertPartialFailure && wasPrimary {
				// enable primary-side semi-sync again
{'fraction_non_alphanumeric': 0.10617626648161, 'fraction_numerical': 0.0006939625260235947, 'mean_word_length': 4.218801871133243, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6979110', 'n_tokens_mistral': 11435, 'n_tokens_neox': 10978, 'n_words': 3526}