Method PurgeDeletes

kv.go:805–872

PurgeDeletes will remove all current delete markers. This is a maintenance option if there is a larger buildup of delete markers. See DeleteMarkersOlderThan() option for more information.

(opts ...PurgeOpt)

// This is a maintenance option if there is a larger buildup of delete markers.
// See DeleteMarkersOlderThan() option for more information.
func (kv *kvs) PurgeDeletes(opts ...PurgeOpt) error {
	var o purgeOpts
	for _, opt := range opts {
		if opt != nil {
			if err := opt.configurePurge(&o); err != nil {
				return err
			}
		}
	}
	// Transfer possible context purge option to the watcher. This is the
	// only option that matters for the PurgeDeletes() feature.
	var wopts []WatchOpt
	if o.ctx != nil {
		wopts = append(wopts, Context(o.ctx))
	}
	watcher, err := kv.WatchAll(wopts...)
	if err != nil {
		return err
	}
	defer watcher.Stop()

	var limit time.Time
	olderThan := o.dmthr
	// Negative value is used to instruct to always remove markers, regardless
	// of age. If set to 0 (or not set), use our default value.
	if olderThan == 0 {
		olderThan = kvDefaultPurgeDeletesMarkerThreshold
	}
	if olderThan > 0 {
		limit = time.Now().Add(-olderThan)
	}

	var deleteMarkers []KeyValueEntry
	for entry := range watcher.Updates() {
		if entry == nil {
			break
		}
		if op := entry.Operation(); op == KeyValueDelete || op == KeyValuePurge {
			deleteMarkers = append(deleteMarkers, entry)
		}
	}
	// Stop watcher here so as we purge we do not have the system continually updating numPending.
	watcher.Stop()

	var (
		pr StreamPurgeRequest
		b  strings.Builder
	)
	// Do actual purges here.
	purgeOpts := []JSOpt{}
	if o.ctx != nil {
		purgeOpts = append(purgeOpts, Context(o.ctx))
	}
	for _, entry := range deleteMarkers {
		b.WriteString(kv.pre)
		b.WriteString(entry.Key())
		pr.Subject = b.String()
		pr.Keep = 0