Method mergeValueForKey

kv/memberlist/memberlist_client.go:1744–1820

Merges incoming value with value we have in our store. Returns "a change" that can be sent to other cluster members to update their state, and new version of the value. If CAS version is specified, then merging will fail if state has changed already, and errVersionMismatch is reported. If no modific

(key string, incomingValue Mergeable, incomingValueRequiresClone bool, casVersion uint, codecID string, deleted bool, updateTime time.Time)

// If CAS version is specified, then merging will fail if state has changed already, and errVersionMismatch is reported.
// If no modification occurred, new version is 0.
func (m *KV) mergeValueForKey(key string, incomingValue Mergeable, incomingValueRequiresClone bool, casVersion uint, codecID string, deleted bool, updateTime time.Time) (change Mergeable, newVersion uint, newDeleted bool, newUpdated time.Time, err error) {
	m.storeMu.Lock()
	defer m.storeMu.Unlock()

	// Note that we do not take a deep copy of curr.value here, it is modified in-place.
	// This is safe because the entire function runs under the store lock; we do not return
	// the full state anywhere as is done elsewhere (i.e. Get/WatchKey/CAS).
	curr := m.store[key]

	// if current entry is nil but the incoming for that key is deleted then we return no change, as we do not want to revive the entry.
	if curr.value == nil && deleted {
		return nil, 0, false, time.Time{}, err
	}

	// if casVersion is 0, then there was no previous value, so we will just do normal merge, without localCAS flag set.
	if casVersion > 0 && curr.Version != casVersion {
		return nil, 0, false, time.Time{}, errVersionMismatch
	}
	result, change, err := computeNewValue(incomingValue, incomingValueRequiresClone, curr.value, casVersion > 0)
	if err != nil {
		return nil, 0, false, time.Time{}, err
	}
	newUpdated = curr.UpdateTime
	newDeleted = curr.Deleted

	// If incoming value is newer, use its timestamp and deleted value
	if !updateTime.IsZero() && updateTime.After(newUpdated) && deleted {
		newUpdated = updateTime
		newDeleted = deleted
	}

	// No change, don't store it.
	if (change == nil || len(change.MergeContent()) == 0) && curr.Deleted == newDeleted {
		return nil, 0, curr.Deleted, curr.UpdateTime, nil
	}

	if m.cfg.LeftIngestersTimeout > 0 {
		limit := time.Now().Add(-m.cfg.LeftIngestersTimeout)
		total, removed := result.RemoveTombstones(limit)
		m.storeTombstones.WithLabelValues(key).Set(float64(total))
		m.storeRemovedTombstones.WithLabelValues(key).Add(float64(removed))

		// Remove tombstones from change too. If change turns out to be empty after this,
		// we don't need to gossip the change. However, the local value will be always be updated.
		//
		// Note that "result" and "change" may actually be the same Mergeable. That is why we
		// call RemoveTombstones on "result" first, so that we get the correct metrics. Calling
		// RemoveTombstones twice with same limit should be noop.
		if change != nil {
			change.RemoveTombstones(limit)
			if len(change.MergeContent()) == 0 {
				return nil, 0, curr.Deleted, curr.UpdateTime, nil
			}
		}
	}

	if change == nil && curr.Deleted != newDeleted {
		// return result as change if the only thing that changes is the Delete state of the entry.