<commit_before>// Copyright 2016, Google Inc. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package throttler

import (
	"fmt"
	"math"
	"sync"
	"time"

	"github.com/youtube/vitess/go/sync2"
	"github.com/youtube/vitess/go/vt/discovery"
	"github.com/youtube/vitess/go/vt/proto/throttlerdata"
	"github.com/youtube/vitess/go/vt/topo/topoproto"

	log "github.com/golang/glog"
	"github.com/golang/protobuf/proto"
)

type state string

const (
	stateIncreaseRate         state = "I"
	stateDecreaseAndGuessRate       = "D"
	stateEmergency                  = "E"
)

type replicationLagChange int

const (
	unknown replicationLagChange = iota
	less
	equal
	greater
)

// replicaUnderTest refers to the replica to which we're currently "locked in"
// i.e. we'll ignore lag records with lower lag from other replicas while we're
// waiting for the next record of this replica under test.
type replicaUnderTest struct {
	// key holds the discovery.TabletStats.Key value for the replica.
	key   string
	alias string
	// state is the "state under test" which triggered the rate change.
	state             state
	nextAllowedChange time.Time
}

// memoryGranularity is the granularity at which values are rounded down in the
// memory instance. This is done to bucket similar rates (see also memory.go).
const memoryGranularity = 5

// MaxReplicationLagModule calculates the maximum rate based on observed
// replication lag and throttler rate changes.
// It implements the Module interface.
// Unless specified, the fields below are not guarded by a Mutex because they
// are only accessed within the Go routine running ProcessRecords().
type MaxReplicationLagModule struct {
	// config holds all parameters for this module.
	// It is only accessed by the Go routine which runs ProcessRecords() and does
	// not require locking.
	config MaxReplicationLagModuleConfig

	// initialMaxReplicationLagSec is the initial value of
	// config.MaxReplicationLagSec with which the module was started. We remember
	// it in case we need to reset the configuration.
	initialMaxReplicationLagSec int64

	// mutableConfigMu guards all fields in the group below.
	mutableConfigMu sync.Mutex
	// mutableConfig is the mutable copy of "config" which is currently used by
	// the module. By modifying "mutableConfig" and setting "applyMutableConfig"
	// to true, the next ProcessRecords() execution will copy "mutableConfig"
	// to "config" and become effective in the module.
	mutableConfig MaxReplicationLagModuleConfig
	// applyMutableConfig is set to true when ProcessRecords() should copy over
	// "mutableConfig" to "config".
	applyMutableConfig bool

	// rate is the rate calculated for the throttler.
	rate         sync2.AtomicInt64
	currentState state
	// lastRateChange is the time when rate was adjusted last.
	lastRateChange             time.Time
	replicaUnderTest           *replicaUnderTest
	nextAllowedChangeAfterInit time.Time

	actualRatesHistory *aggregatedIntervalHistory
	lagCache           *replicationLagCache
	// memory tracks known good and bad throttler rates.
	memory *memory

	// rateUpdateChan is the notification channel to tell the throttler when our
	// max rate calculation has changed. The field is immutable (set in Start().)
	rateUpdateChan chan<- struct{}

	// lagRecords buffers the replication lag records received by the HealthCheck
	// listener. ProcessRecords() will process them.
	lagRecords chan replicationLagRecord
	wg         sync.WaitGroup

	// results caches the results of the latest processed replication lag records.
	results *resultRing
}

// NewMaxReplicationLagModule will create a new module instance and set the
// initial max replication lag limit to maxReplicationLag.
func NewMaxReplicationLagModule(config MaxReplicationLagModuleConfig, actualRatesHistory *aggregatedIntervalHistory, nowFunc func() time.Time) (*MaxReplicationLagModule, error) {
	if err := config.Verify(); err != nil {
		return nil, fmt.Errorf("invalid NewMaxReplicationLagModuleConfig: %v", err)
	}
	rate := int64(ReplicationLagModuleDisabled)
	if config.MaxReplicationLagSec != ReplicationLagModuleDisabled {
		rate = config.InitialRate
	}

	m := &MaxReplicationLagModule{
		initialMaxReplicationLagSec: config.MaxReplicationLagSec,
		// Register "config" for a future config update.
		mutableConfig:      config,
		applyMutableConfig: true,
		// Always start off with a non-zero rate because zero means all requests
		// get throttled.
		rate:           sync2.NewAtomicInt64(rate),
		currentState:   stateIncreaseRate,
		lastRateChange: nowFunc(),
		memory:         newMemory(memoryGranularity, config.AgeBadRateAfter(), config.BadRateIncrease),
		lagRecords:     make(chan replicationLagRecord, 10),
		// Prevent an immediate increase of the initial rate.
		nextAllowedChangeAfterInit: nowFunc().Add(config.MaxDurationBetweenIncreases()),
		actualRatesHistory:         actualRatesHistory,
		lagCache:                   newReplicationLagCache(1000),
		results:                    newResultRing(1000),
	}

	// Enforce a config update.
	m.applyLatestConfig()

	return m, nil
}

// Start launches a Go routine which reacts on replication lag records.
// It implements the Module interface.
func (m *MaxReplicationLagModule) Start(rateUpdateChan chan<- struct{}) {
	m.rateUpdateChan = rateUpdateChan
	m.wg.Add(1)
	go m.ProcessRecords()
}

// Stop blocks until the module's Go routine is stopped.
// It implements the Module interface.
func (m *MaxReplicationLagModule) Stop() {
	close(m.lagRecords)
	m.wg.Wait()
}

// MaxRate returns the current maximum allowed rate.
// It implements the Module interface.
func (m *MaxReplicationLagModule) MaxRate() int64 {
	return m.rate.Get()
}

// applyLatestConfig checks if "mutableConfig" should be applied as the new
// "config" and does so when necessary.
func (m *MaxReplicationLagModule) applyLatestConfig() {
	var config MaxReplicationLagModuleConfig
	applyConfig := false

	m.mutableConfigMu.Lock()
	if m.applyMutableConfig {
		config = m.mutableConfig
		applyConfig = true
		m.applyMutableConfig = false
	}
	m.mutableConfigMu.Unlock()

	// No locking required here because this method is only called from the same
	// Go routine as ProcessRecords() or the constructor.
	if applyConfig {
		m.config = config
		m.memory.updateAgingConfiguration(config.AgeBadRateAfter(), config.BadRateIncrease)
	}
}

func (m *MaxReplicationLagModule) getConfiguration() *throttlerdata.Configuration {
	m.mutableConfigMu.Lock()
	defer m.mutableConfigMu.Unlock()

	configCopy := m.mutableConfig.Configuration
	return &configCopy
}

func (m *MaxReplicationLagModule) updateConfiguration(configuration *throttlerdata.Configuration, copyZeroValues bool) error {
	m.mutableConfigMu.Lock()
	defer m.mutableConfigMu.Unlock()

	newConfig := m.mutableConfig

	if copyZeroValues {
		newConfig.Configuration = *proto.Clone(configuration).(*throttlerdata.Configuration)
	} else {
		proto.Merge(&newConfig.Configuration, configuration)
	}

	if err := newConfig.Verify(); err != nil {
		return err
	}
	m.mutableConfig = newConfig
	m.applyMutableConfig = true
	return nil
}

func (m *MaxReplicationLagModule) resetConfiguration() {
	m.mutableConfigMu.Lock()
	defer m.mutableConfigMu.Unlock()

	m.mutableConfig = NewMaxReplicationLagModuleConfig(m.initialMaxReplicationLagSec)
	m.applyMutableConfig = true
}

// RecordReplicationLag records the current replication lag for processing.
func (m *MaxReplicationLagModule) RecordReplicationLag(t time.Time, ts *discovery.TabletStats) {
	m.mutableConfigMu.Lock()
	if m.mutableConfig.MaxReplicationLagSec == ReplicationLagModuleDisabled {
		m.mutableConfigMu.Unlock()
		return
	}
	m.mutableConfigMu.Unlock()

	// Buffer data point for now to unblock the HealthCheck listener and process
	// it asynchronously in ProcessRecords().
	m.lagRecords <- replicationLagRecord{t, *ts}
}

// ProcessRecords is the main loop, run in a separate Go routine, which
// reacts to any replication lag updates (recordings).
func (m *MaxReplicationLagModule) ProcessRecords() {
	defer m.wg.Done()

	for lagRecord := range m.lagRecords {
		m.processRecord(lagRecord)
	}
}

func (m *MaxReplicationLagModule) processRecord(lagRecord replicationLagRecord) {
	m.applyLatestConfig()

	m.lagCache.add(lagRecord)

	m.lagCache.sortByLag(int(m.config.IgnoreNSlowestReplicas), m.config.MaxReplicationLagSec+1)

	m.recalculateRate(lagRecord)
}

func (m *MaxReplicationLagModule) recalculateRate(lagRecordNow replicationLagRecord) {
	if lagRecordNow.isZero() {
		panic("rate recalculation was triggered with a zero replication lag record")
	}
	now := lagRecordNow.time
	lagNow := lagRecordNow.lag()

	m.memory.ageBadRate(now)

	r := result{
		Now:            now,
		RateChange:     unchangedRate,
		lastRateChange: m.lastRateChange,
		OldState:       m.currentState,
		NewState:       m.currentState,
		OldRate:        m.rate.Get(),
		NewRate:        m.rate.Get(),
		LagRecordNow:   lagRecordNow,
	}
	if lagNow <= m.config.TargetReplicationLagSec {
		// Lag in range: [0, target]
		r.TestedState = stateIncreaseRate
	} else if lagNow <= m.config.MaxReplicationLagSec {
		// Lag in range: (target, max]
		r.TestedState = stateDecreaseAndGuessRate
	} else {
		// Lag in range: (max, infinite]
		r.TestedState = stateEmergency
	}

	// Declare new variables before we use "goto". Required by the Go compiler.
	var clear bool
	var clearReason string

	if m.lagCache.ignoreSlowReplica(lagRecordNow.Key) {
		r.Reason = fmt.Sprintf("skipping this replica because it's among the %d slowest replicas", m.config.IgnoreNSlowestReplicas)
		goto logResult
	}

	clear, clearReason = m.clearReplicaUnderTest(now, r.TestedState, lagRecordNow)
	if clear {
		clearReason = fmt.Sprintf("; previous replica under test (%v) cleared because %v", m.replicaUnderTest.alias, clearReason)
		m.replicaUnderTest = nil
	}

	if r.TestedState == stateIncreaseRate && now.Before(m.nextAllowedChangeAfterInit) {
		r.Reason = fmt.Sprintf("waiting %.1f more seconds since init until we try the first increase", m.nextAllowedChangeAfterInit.Sub(now).Seconds())
		goto logResult
	}

	if !m.isReplicaUnderTest(&r, now, r.TestedState, lagRecordNow) {
		goto logResult
	}

	// Process the lag record and adjust the rate.
	if m.replicaUnderTest != nil {
		// We're checking the same replica again. The old value is no longer needed.
		m.replicaUnderTest = nil
	}
	switch r.TestedState {
	case stateIncreaseRate:
		m.increaseRate(&r, now, lagRecordNow)
	case stateDecreaseAndGuessRate:
		m.decreaseAndGuessRate(&r, now, lagRecordNow)
	case stateEmergency:
		m.emergency(&r, now, lagRecordNow)
	default:
		panic(fmt.Sprintf("BUG: invalid state: %v", r.TestedState))
	}

logResult:
	r.HighestGood = m.memory.highestGood()
	r.LowestBad = m.memory.lowestBad()

	if clear {
		r.Reason += clearReason
	}

	log.Infof("%v", r)
	m.results.add(r)
}

// clearReplicaUnderTest returns true if the current "replica under test" should
// be cleared e.g. because the new lag record is more severe or we did not hear
// back from the replica under test for a while.
func (m *MaxReplicationLagModule) clearReplicaUnderTest(now time.Time, testedState state, lagRecordNow replicationLagRecord) (bool, string) {
	if m.replicaUnderTest == nil {
		return false, ""
	}

	if stateGreater(testedState, m.replicaUnderTest.state) {
		// Example: "decrease" > "increase" will return true.
		return true, fmt.Sprintf("the new state is more severe (%v -> %v)", m.replicaUnderTest.state, testedState)
	}

	// Verify that the current replica under test is not in an error state.
	lr := lagRecordNow
	if m.replicaUnderTest.key != lr.Key {
		lr = m.lagCache.latest(m.replicaUnderTest.key)
	}
	if lr.isZero() {
		// Replica is no longer tracked by the cache i.e. may be offline.
		return true, "it is no longer actively tracked"
	}
	if lr.LastError != nil {
		// LastError is set i.e. HealthCheck module cannot connect and the cached
		// data for the replica might be outdated.
		return true, "it has LastError set i.e. is no longer correctly tracked"
	}

	if m.lagCache.isIgnored(m.replicaUnderTest.key) {
		// "replica under test" has become a slow, ignored replica.
		return true, "its ignored as a slow replica"
	}

	if now.After(m.replicaUnderTest.nextAllowedChange.Add(m.config.MaxDurationBetweenIncreases())) {
		// We haven't heard from the replica under test for too long. Assume it did
		// timeout.
		return true, fmt.Sprintf("we didn't see a recent record from it within the last %.1f seconds", m.config.MaxDurationBetweenIncreases().Seconds())
	}

	return false, ""
}

// stateGreater returns true if a > b i.e. the state "a" is more severe than
// "b". For example, "decrease" > "increase" returns true.
func stateGreater(a, b state) bool {
	switch a {
	case stateIncreaseRate:
		return false
	case stateDecreaseAndGuessRate:
		return b == stateIncreaseRate
	case stateEmergency:
		return b == stateIncreaseRate || b == stateDecreaseAndGuessRate
	default:
		panic(fmt.Sprintf("BUG: cannot compare states: %v and %v", a, b))
	}
}

// isReplicaUnderTest returns true if a 'replica under test' is currently set<commit_msg>throttler: Fix typo.<commit_after>// Copyright 2016, Google Inc. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package throttler

import (
	"fmt"
	"math"
	"sync"
	"time"

	"github.com/youtube/vitess/go/sync2"
	"github.com/youtube/vitess/go/vt/discovery"
	"github.com/youtube/vitess/go/vt/proto/throttlerdata"
	"github.com/youtube/vitess/go/vt/topo/topoproto"

	log "github.com/golang/glog"
	"github.com/golang/protobuf/proto"
)

type state string

const (
	stateIncreaseRate         state = "I"
	stateDecreaseAndGuessRate       = "D"
	stateEmergency                  = "E"
)

type replicationLagChange int

const (
	unknown replicationLagChange = iota
	less
	equal
	greater
)

// replicaUnderTest refers to the replica to which we're currently "locked in"
// i.e. we'll ignore lag records with lower lag from other replicas while we're
// waiting for the next record of this replica under test.
type replicaUnderTest struct {
	// key holds the discovery.TabletStats.Key value for the replica.
	key   string
	alias string
	// state is the "state under test" which triggered the rate change.
	state             state
	nextAllowedChange time.Time
}

// memoryGranularity is the granularity at which values are rounded down in the
// memory instance. This is done to bucket similar rates (see also memory.go).
const memoryGranularity = 5

// MaxReplicationLagModule calculates the maximum rate based on observed
// replication lag and throttler rate changes.
// It implements the Module interface.
// Unless specified, the fields below are not guarded by a Mutex because they
// are only accessed within the Go routine running ProcessRecords().
type MaxReplicationLagModule struct {
	// config holds all parameters for this module.
	// It is only accessed by the Go routine which runs ProcessRecords() and does
	// not require locking.
	config MaxReplicationLagModuleConfig

	// initialMaxReplicationLagSec is the initial value of
	// config.MaxReplicationLagSec with which the module was started. We remember
	// it in case we need to reset the configuration.
	initialMaxReplicationLagSec int64

	// mutableConfigMu guards all fields in the group below.
	mutableConfigMu sync.Mutex
	// mutableConfig is the mutable copy of "config" which is currently used by
	// the module. By modifying "mutableConfig" and setting "applyMutableConfig"
	// to true, the next ProcessRecords() execution will copy "mutableConfig"
	// to "config" and become effective in the module.
	mutableConfig MaxReplicationLagModuleConfig
	// applyMutableConfig is set to true when ProcessRecords() should copy over
	// "mutableConfig" to "config".
	applyMutableConfig bool

	// rate is the rate calculated for the throttler.
	rate         sync2.AtomicInt64
	currentState state
	// lastRateChange is the time when rate was adjusted last.
	lastRateChange             time.Time
	replicaUnderTest           *replicaUnderTest
	nextAllowedChangeAfterInit time.Time

	actualRatesHistory *aggregatedIntervalHistory
	lagCache           *replicationLagCache
	// memory tracks known good and bad throttler rates.
	memory *memory

	// rateUpdateChan is the notification channel to tell the throttler when our
	// max rate calculation has changed. The field is immutable (set in Start().)
	rateUpdateChan chan<- struct{}

	// lagRecords buffers the replication lag records received by the HealthCheck
	// listener. ProcessRecords() will process them.
	lagRecords chan replicationLagRecord
	wg         sync.WaitGroup

	// results caches the results of the latest processed replication lag records.
	results *resultRing
}

// NewMaxReplicationLagModule will create a new module instance and set the
// initial max replication lag limit to maxReplicationLag.
func NewMaxReplicationLagModule(config MaxReplicationLagModuleConfig, actualRatesHistory *aggregatedIntervalHistory, nowFunc func() time.Time) (*MaxReplicationLagModule, error) {
	if err := config.Verify(); err != nil {
		return nil, fmt.Errorf("invalid NewMaxReplicationLagModuleConfig: %v", err)
	}
	rate := int64(ReplicationLagModuleDisabled)
	if config.MaxReplicationLagSec != ReplicationLagModuleDisabled {
		rate = config.InitialRate
	}

	m := &MaxReplicationLagModule{
		initialMaxReplicationLagSec: config.MaxReplicationLagSec,
		// Register "config" for a future config update.
		mutableConfig:      config,
		applyMutableConfig: true,
		// Always start off with a non-zero rate because zero means all requests
		// get throttled.
		rate:           sync2.NewAtomicInt64(rate),
		currentState:   stateIncreaseRate,
		lastRateChange: nowFunc(),
		memory:         newMemory(memoryGranularity, config.AgeBadRateAfter(), config.BadRateIncrease),
		lagRecords:     make(chan replicationLagRecord, 10),
		// Prevent an immediate increase of the initial rate.
		nextAllowedChangeAfterInit: nowFunc().Add(config.MaxDurationBetweenIncreases()),
		actualRatesHistory:         actualRatesHistory,
		lagCache:                   newReplicationLagCache(1000),
		results:                    newResultRing(1000),
	}

	// Enforce a config update.
	m.applyLatestConfig()

	return m, nil
}

// Start launches a Go routine which reacts on replication lag records.
// It implements the Module interface.
func (m *MaxReplicationLagModule) Start(rateUpdateChan chan<- struct{}) {
	m.rateUpdateChan = rateUpdateChan
	m.wg.Add(1)
	go m.ProcessRecords()
}

// Stop blocks until the module's Go routine is stopped.
// It implements the Module interface.
func (m *MaxReplicationLagModule) Stop() {
	close(m.lagRecords)
	m.wg.Wait()
}

// MaxRate returns the current maximum allowed rate.
// It implements the Module interface.
func (m *MaxReplicationLagModule) MaxRate() int64 {
	return m.rate.Get()
}

// applyLatestConfig checks if "mutableConfig" should be applied as the new
// "config" and does so when necessary.
func (m *MaxReplicationLagModule) applyLatestConfig() {
	var config MaxReplicationLagModuleConfig
	applyConfig := false

	m.mutableConfigMu.Lock()
	if m.applyMutableConfig {
		config = m.mutableConfig
		applyConfig = true
		m.applyMutableConfig = false
	}
	m.mutableConfigMu.Unlock()

	// No locking required here because this method is only called from the same
	// Go routine as ProcessRecords() or the constructor.
	if applyConfig {
		m.config = config
		m.memory.updateAgingConfiguration(config.AgeBadRateAfter(), config.BadRateIncrease)
	}
}

func (m *MaxReplicationLagModule) getConfiguration() *throttlerdata.Configuration {
	m.mutableConfigMu.Lock()
	defer m.mutableConfigMu.Unlock()

	configCopy := m.mutableConfig.Configuration
	return &configCopy
}

func (m *MaxReplicationLagModule) updateConfiguration(configuration *throttlerdata.Configuration, copyZeroValues bool) error {
	m.mutableConfigMu.Lock()
	defer m.mutableConfigMu.Unlock()

	newConfig := m.mutableConfig

	if copyZeroValues {
		newConfig.Configuration = *proto.Clone(configuration).(*throttlerdata.Configuration)
	} else {
		proto.Merge(&newConfig.Configuration, configuration)
	}

	if err := newConfig.Verify(); err != nil {
		return err
	}
	m.mutableConfig = newConfig
	m.applyMutableConfig = true
	return nil
}

func (m *MaxReplicationLagModule) resetConfiguration() {
	m.mutableConfigMu.Lock()
	defer m.mutableConfigMu.Unlock()

	m.mutableConfig = NewMaxReplicationLagModuleConfig(m.initialMaxReplicationLagSec)
	m.applyMutableConfig = true
}

// RecordReplicationLag records the current replication lag for processing.
func (m *MaxReplicationLagModule) RecordReplicationLag(t time.Time, ts *discovery.TabletStats) {
	m.mutableConfigMu.Lock()
	if m.mutableConfig.MaxReplicationLagSec == ReplicationLagModuleDisabled {
		m.mutableConfigMu.Unlock()
		return
	}
	m.mutableConfigMu.Unlock()

	// Buffer data point for now to unblock the HealthCheck listener and process
	// it asynchronously in ProcessRecords().
	m.lagRecords <- replicationLagRecord{t, *ts}
}

// ProcessRecords is the main loop, run in a separate Go routine, which
// reacts to any replication lag updates (recordings).
func (m *MaxReplicationLagModule) ProcessRecords() {
	defer m.wg.Done()

	for lagRecord := range m.lagRecords {
		m.processRecord(lagRecord)
	}
}

func (m *MaxReplicationLagModule) processRecord(lagRecord replicationLagRecord) {
	m.applyLatestConfig()

	m.lagCache.add(lagRecord)

	m.lagCache.sortByLag(int(m.config.IgnoreNSlowestReplicas), m.config.MaxReplicationLagSec+1)

	m.recalculateRate(lagRecord)
}

func (m *MaxReplicationLagModule) recalculateRate(lagRecordNow replicationLagRecord) {
	if lagRecordNow.isZero() {
		panic("rate recalculation was triggered with a zero replication lag record")
	}
	now := lagRecordNow.time
	lagNow := lagRecordNow.lag()

	m.memory.ageBadRate(now)

	r := result{
		Now:            now,
		RateChange:     unchangedRate,
		lastRateChange: m.lastRateChange,
		OldState:       m.currentState,
		NewState:       m.currentState,
		OldRate:        m.rate.Get(),
		NewRate:        m.rate.Get(),
		LagRecordNow:   lagRecordNow,
	}
	if lagNow <= m.config.TargetReplicationLagSec {
		// Lag in range: [0, target]
		r.TestedState = stateIncreaseRate
	} else if lagNow <= m.config.MaxReplicationLagSec {
		// Lag in range: (target, max]
		r.TestedState = stateDecreaseAndGuessRate
	} else {
		// Lag in range: (max, infinite]
		r.TestedState = stateEmergency
	}

	// Declare new variables before we use "goto". Required by the Go compiler.
	var clear bool
	var clearReason string

	if m.lagCache.ignoreSlowReplica(lagRecordNow.Key) {
		r.Reason = fmt.Sprintf("skipping this replica because it's among the %d slowest replicas", m.config.IgnoreNSlowestReplicas)
		goto logResult
	}

	clear, clearReason = m.clearReplicaUnderTest(now, r.TestedState, lagRecordNow)
	if clear {
		clearReason = fmt.Sprintf("; previous replica under test (%v) cleared because %v", m.replicaUnderTest.alias, clearReason)
		m.replicaUnderTest = nil
	}

	if r.TestedState == stateIncreaseRate && now.Before(m.nextAllowedChangeAfterInit) {
		r.Reason = fmt.Sprintf("waiting %.1f more seconds since init until we try the first increase", m.nextAllowedChangeAfterInit.Sub(now).Seconds())
		goto logResult
	}

	if !m.isReplicaUnderTest(&r, now, r.TestedState, lagRecordNow) {
		goto logResult
	}

	// Process the lag record and adjust the rate.
	if m.replicaUnderTest != nil {
		// We're checking the same replica again. The old value is no longer needed.
		m.replicaUnderTest = nil
	}
	switch r.TestedState {
	case stateIncreaseRate:
		m.increaseRate(&r, now, lagRecordNow)
	case stateDecreaseAndGuessRate:
		m.decreaseAndGuessRate(&r, now, lagRecordNow)
	case stateEmergency:
		m.emergency(&r, now, lagRecordNow)
	default:
		panic(fmt.Sprintf("BUG: invalid state: %v", r.TestedState))
	}

logResult:
	r.HighestGood = m.memory.highestGood()
	r.LowestBad = m.memory.lowestBad()

	if clear {
		r.Reason += clearReason
	}

	log.Infof("%v", r)
	m.results.add(r)
}

// clearReplicaUnderTest returns true if the current "replica under test" should
// be cleared e.g. because the new lag record is more severe or we did not hear
// back from the replica under test for a while.
func (m *MaxReplicationLagModule) clearReplicaUnderTest(now time.Time, testedState state, lagRecordNow replicationLagRecord) (bool, string) {
	if m.replicaUnderTest == nil {
		return false, ""
	}

	if stateGreater(testedState, m.replicaUnderTest.state) {
		// Example: "decrease" > "increase" will return true.
		return true, fmt.Sprintf("the new state is more severe (%v -> %v)", m.replicaUnderTest.state, testedState)
	}

	// Verify that the current replica under test is not in an error state.
	lr := lagRecordNow
	if m.replicaUnderTest.key != lr.Key {
		lr = m.lagCache.latest(m.replicaUnderTest.key)
	}
	if lr.isZero() {
		// Replica is no longer tracked by the cache i.e. may be offline.
		return true, "it is no longer actively tracked"
	}
	if lr.LastError != nil {
		// LastError is set i.e. HealthCheck module cannot connect and the cached
		// data for the replica might be outdated.
		return true, "it has LastError set i.e. is no longer correctly tracked"
	}

	if m.lagCache.isIgnored(m.replicaUnderTest.key) {
		// "replica under test" has become a slow, ignored replica.
		return true, "it is ignored as a slow replica"
	}

	if now.After(m.replicaUnderTest.nextAllowedChange.Add(m.config.MaxDurationBetweenIncreases())) {
		// We haven't heard from the replica under test for too long. Assume it did
		// timeout.
		return true, fmt.Sprintf("we didn't see a recent record from it within the last %.1f seconds", m.config.MaxDurationBetweenIncreases().Seconds())
	}

	return false, ""
}

// stateGreater returns true if a > b i.e. the state "a" is more severe than
// "b". For example, "decrease" > "increase" returns true.
func stateGreater(a, b state) bool {
	switch a {
	case stateIncreaseRate:
		return false
	case stateDecreaseAndGuessRate:
		return b == stateIncreaseRate
	case stateEmergency:
		return b == stateIncreaseRate || b == stateDecreaseAndGuessRate
	default:
		panic(fmt.Sprintf("BUG: cannot compare states: %v and %v", a, b))
	}
}

// isReplicaUnderTest returns true if a 'replica under test' is currently set
{'fraction_non_alphanumeric': 0.08983469504085123, 'fraction_numerical': 0.002508075242257268, 'mean_word_length': 4.128824790489183, 'pattern_counts': {'":': 0, '<': 13, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '4323495', 'n_tokens_mistral': 8446, 'n_tokens_neox': 7903, 'n_words': 2538}