buffer_delete.go

// Delete paths on *scopeBuffer:
//
//   - deleteByID     — single-item delete by id
//   - deleteBySeq    — single-item delete by seq
//   - trimSeq  — drain the prefix [seq=1 .. seq=maxSeq] in one shot
//
// All three take b.mu exclusively, check b.detached first, and route
// byte releases through the store-wide totalBytes counter. The
// low-level helper deleteIndexLocked centralises the GC-zeroing,
// secondary-index sync, and counter update so the three callers cannot
// drift.

package scopecache

import "sort"

// deleteIndexLocked removes items[i] in O(n) tail-shift, nils the
// now-duplicate last slot, removes the byID entry (when present), and
// releases the item's bytes from b.bytes and (when store-attached)
// s.totalBytes.
//
// PRECONDITION: caller holds b.mu and i is valid.
//
// Three invariants the body upholds — drift here leaks state silently:
//
//  1. Nil the duplicate last slot before reslicing, otherwise the
//     backing array keeps a pointer to the removed *Item (and its
//     payload bytes) and prevents GC.
//  2. b.bytes and s.totalBytes update lockstep.
//  3. byID delete is conditional on removed.ID != "" — empty-id
//     items have no byID entry to remove. Seq lookups need no removal
//     because b.items is the seq index (indexBySeqLocked).
func (b *scopeBuffer) deleteIndexLocked(i int) {
	removed := b.items[i]
	removedSize := approxItemSize(*removed)

	// Tail-shift, then nil the now-duplicate last slot before shrinking
	// (invariant 1 above).
	copy(b.items[i:], b.items[i+1:])
	b.items[len(b.items)-1] = nil
	b.items = b.items[:len(b.items)-1]

	if removed.ID != "" {
		delete(b.byID, removed.ID)
	}

	b.bytes -= removedSize
	now := nowUnixMicro()
	b.store.totalBytes.Add(-removedSize)
	b.store.totalItems.Add(-1)
	b.store.bumpLastWriteTS(now)
	b.bumpLastWriteTSLocked(now)
	b.freeItemLocked(removed)
	b.resetIfEmptyLocked()
	b.shrinkIfSparseLocked()
}

// shrinkIfSparseLocked compacts the scope's storage when it has gone
// substantially sparse — total chunk capacity > shrinkMinCap AND live
// items < capacity/shrinkSparseRatio. A drain leaves three things at the
// historical high-watermark that this reclaims together:
//
//   - the items slice backing array,
//   - the byID map bucket array (Go maps never shrink on delete),
//   - the Item chunk arena AND the freeItemLocked free-list — a drain
//     that shrinks but does not EMPTY a scope (so resetIfEmptyLocked
//     never fires) keeps every freed slot on itemFree, never reused, and
//     the chunks holding them pinned. Keying off chunk capacity (not
//     cap(b.items)) is what catches the bulk /trim path, whose
//     own `rest` rebuild already compacted the items slice.
//
// Live Items are COPIED BY VALUE into a single fresh chunk — payload /
// rawBytes byte slices are shared via the copied headers, never
// duplicated — and items/byID re-point into it; the old chunks +
// free-list drop to GC. Moving Items is safe: every index is rebuilt
// here under b.mu, and reads copy the Item out under the lock, so no
// caller can retain a stale *Item across the move.
//
// Conservative thresholds — only fires on visibly-sparse buffers, not on
// minor churn. After a rebuild capacity == len, so a follow-up delete
// must again accumulate sparsity before the next rebuild; amortised O(1).
//
// PRECONDITION: caller holds b.mu and the delete logic has already
// updated b.bytes / counters.
const (
	shrinkMinCap      = 1024
	shrinkSparseRatio = 4 // live items < capacity/4 → rebuild
)

func (b *scopeBuffer) shrinkIfSparseLocked() {
	// Trigger on the larger of the two retained backing arrays: the
	// items slice (cap(b.items), grown by append-doubling) and the Item
	// chunk arena (itemCapacityLocked). The single-delete path leaves
	// the slice cap at the high-watermark; the bulk /trim path
	// compacts the slice via its `rest` rebuild but leaves the chunks
	// large — taking the max catches both.
	retainedCap := b.itemCapacityLocked()
	if c := cap(b.items); c > retainedCap {
		retainedCap = c
	}
	if retainedCap <= shrinkMinCap {
		return
	}
	if len(b.items)*shrinkSparseRatio >= retainedCap {
		return
	}

	n := len(b.items)
	storage := make([]Item, n)
	newItems := make([]*Item, n)
	var newByID map[string]*Item
	for i := range b.items {
		storage[i] = *b.items[i] // value copy into the fresh chunk
		p := &storage[i]
		newItems[i] = p
		if p.ID != "" {
			if newByID == nil {
				newByID = make(map[string]*Item, n)
			}
			newByID[p.ID] = p
		}
	}
	b.items = newItems
	b.byID = newByID
	b.itemChunks = [][]Item{storage}
	b.chunkUsed = n
	b.itemFree = nil
}

// resetIfEmptyLocked may drop large high-watermark backing storage
// after a scope has just been drained to zero items. Without it,
// drained scopes retain the slice's full backing array and the maps'
// bucket arrays (Go maps don't shrink on delete) until the next write
// grows them. Hysteresis (see below) keeps the backing arrays of
// scopes that never grew past itemChunkReclaimThreshold.
//
// nil-ing is safe: write paths lazy-init the maps on first write
// after a reset, and append() on a nil slice grows naturally.
// b.lastSeq is intentionally not reset — the seq cursor must stay
// monotonic across drain/refill cycles so downstream consumers
// tracking it cannot observe a regression.
//
// PRECONDITION: caller holds b.mu and the delete that produced the
// empty state has already updated b.bytes / counters.
func (b *scopeBuffer) resetIfEmptyLocked() {
	if len(b.items) != 0 {
		return
	}
	// Hysteresis: a scope that never grew past itemChunkReclaimThreshold
	// keeps its (small) backing arrays, chunks and freeList so a churn /
	// drain-stream refill reuses them at zero allocations. Only a scope
	// that grew large returns its memory on full drain.
	if b.itemCapacityLocked() <= itemChunkReclaimThreshold {
		return
	}
	b.items = nil
	b.byID = nil
	b.itemChunks = nil
	b.itemFree = nil
	b.chunkUsed = 0
}

func (b *scopeBuffer) deleteByID(id string) (int, error) {
	b.mu.Lock()
	defer b.mu.Unlock()

	if b.detached {
		return 0, &ScopeDetachedError{}
	}

	existing, ok := b.byID[id]
	if !ok {
		return 0, nil
	}

	i, ok := b.indexBySeqLocked(existing.Seq)
	if !ok {
		// Defensive: byID and items stay in sync under b.mu.
		return 0, nil
	}
	b.deleteIndexLocked(i)
	return 1, nil
}

func (b *scopeBuffer) deleteBySeq(seq uint64) (int, error) {
	b.mu.Lock()
	defer b.mu.Unlock()

	if b.detached {
		return 0, &ScopeDetachedError{}
	}

	i, ok := b.indexBySeqLocked(seq)
	if !ok {
		return 0, nil
	}
	b.deleteIndexLocked(i)
	return 1, nil
}

// trimSeq removes every item with Seq <= maxSeq. b.items is
// always ordered ascending by Seq — appendItem assigns monotonic
// seqs, and the delete paths preserve relative order of the
// remaining items — so binary search finds the cut point in
// O(log n). Returns the number of items removed and any
// *ScopeDetachedError if the buffer was orphaned by /drop_scope,
// /flush, or /rebuild before the caller's mutation could land.
func (b *scopeBuffer) trimSeq(maxSeq uint64) (int, error) {
	b.mu.Lock()
	defer b.mu.Unlock()

	if b.detached {
		return 0, &ScopeDetachedError{}
	}

	idx := sort.Search(len(b.items), func(i int) bool {
		return b.items[i].Seq > maxSeq
	})
	if idx == 0 {
		return 0, nil
	}

	var freedBytes int64
	for i := 0; i < idx; i++ {
		removed := b.items[i]
		freedBytes += approxItemSize(*removed)
		if removed.ID != "" {
			delete(b.byID, removed.ID)
		}
		b.freeItemLocked(removed)
	}
	// Copy the kept suffix into a fresh backing array so the old one —
	// which still holds the removed payloads in its prefix — becomes
	// GC-eligible. A bare reslice (b.items[idx:]) would pin the full
	// original array behind a small remainder; this matters for the
	// write-buffer pattern where repeated drain-from-front otherwise
	// retains memory proportional to the historical high-watermark.
	rest := make([]*Item, len(b.items)-idx)
	copy(rest, b.items[idx:])
	b.items = rest

	b.bytes -= freedBytes
	now := nowUnixMicro()
	b.store.totalBytes.Add(-freedBytes)
	b.store.totalItems.Add(-int64(idx))
	b.store.bumpLastWriteTS(now)
	b.bumpLastWriteTSLocked(now)
	// `rest` already freed the items-slice backing array. resetIfEmptyLocked
	// reclaims everything when this drain emptied the scope; otherwise
	// shrinkIfSparseLocked reclaims the chunk arena + free-list + byID
	// map buckets when the drain left the scope sparse-but-not-empty
	// (the `rest` rebuild compacted only the items slice, so the
	// chunk-capacity trigger is what catches this path).
	b.resetIfEmptyLocked()
	b.shrinkIfSparseLocked()
	return idx, nil
}