platform/projection/manager.go

// SnapshotHolder provides atomic hot-reload for SnapshotReader and solves the
// classic "Get concurrent with Close/munmap" race that would otherwise produce
// uncatchable SIGBUS / SIGSEGV under reload.
//
// Safety is an RWMutex (readMu): Get holds RLock across the load-and-read of a
// reader, and a reader is only Close()d (munmap) under Lock, so a reader can
// never be unmapped while a Get is reading it — regardless of how long that
// Get's goroutine was descheduled. (A drain-grace timer alone, the original
// approach, is only best-effort: a goroutine paused longer than the grace can
// still read freed pages. TestSnapshotHolder_SwapRace reproduces that crash.)
//
// The drain grace is retained as a close-BATCHING optimization: it defers and
// coalesces reader closes onto the reaper tick instead of taking the exclusive
// Lock on every swap. Pattern source: slask-finder's mmap+reload stores.
//
// Use:
//
//	holder := projection.NewSnapshotHolder[T](30 * time.Second) // drain grace
//	if err := holder.Open("/var/lib/catalog/bygglagret.bin"); err != nil { ... }
//	// later, when a fresh snapshot lands:
//	if err := holder.Reload("/var/lib/catalog/bygglagret.bin"); err != nil { ... }
//	// pulls the value; always hits the current reader:
//	p, ok := holder.Get(key)
//
// holder.Get / Scan / Len are safe for parallel use. Open and Reload are also
// safe for parallel use. Close stops the reaper goroutine and shuts the
// current reader + drains down.
package projection

import (
	"sync"
	"sync/atomic"
	"time"
)

// DefaultDrainGrace is the default drain period for SnapshotHolder. Finder's
// analogous reload path uses ~30s; we mirror that value.
const DefaultDrainGrace = 30 * time.Second

// SnapshotHolder swaps a SnapshotReader under load without unmapping pages
// out from under in-flight Get()s. Set retains the previous reader behind a
// drain timer; after the timer elapses the reaper goroutine closes the old
// reader (and runs its munmap).
type SnapshotHolder[T any] struct {
	current atomic.Pointer[SnapshotReader[T]]

	// readMu makes reclamation race-free: Get holds RLock across the
	// Load-and-read of a reader, and a reader is only Close()d (munmap) under
	// Lock — so a reader can never be unmapped while a Get is reading it, even
	// if that Get's goroutine was descheduled past the drain grace.
	readMu sync.RWMutex

	drainsMu   sync.Mutex
	drains     []*pendingClose[T]
	drainGrace time.Duration

	stopCh chan struct{}
	once   sync.Once
}

type pendingClose[T any] struct {
	r        *SnapshotReader[T]
	expireAt time.Time
}

// NewSnapshotHolder returns a holder of T with the given drain grace (the time
// the previous reader lives behind Set before being closed). Zero or
// negative grace falls back to DefaultDrainGrace.
func NewSnapshotHolder[T any](grace time.Duration) *SnapshotHolder[T] {
	if grace <= 0 {
		grace = DefaultDrainGrace
	}
	h := &SnapshotHolder[T]{
		drainGrace: grace,
		stopCh:     make(chan struct{}),
	}
	go h.sweepLoop()
	return h
}

// Open is shorthand for Set(ReadSnapshotFile(path)). Errors leave the
// holder's current reader unchanged.
func (h *SnapshotHolder[T]) Open(path string) error {
	r, err := ReadSnapshotFile[T](path)
	if err != nil {
		return err
	}
	h.Set(r)
	return nil
}

// Reload atomically swaps the current reader for one freshly opened from
// path. The old reader (if any) is queued for close after the drain grace
// elapses; in-flight Get(s) on the old reader continue to work.
func (h *SnapshotHolder[T]) Reload(path string) error {
	r, err := ReadSnapshotFile[T](path)
	if err != nil {
		return err
	}
	h.Set(r)
	return nil
}

// Set replaces the current reader with r. The previous reader is queued
// for close after the drain grace elapses. r may be nil to clear.
func (h *SnapshotHolder[T]) Set(r *SnapshotReader[T]) {
	if r == nil {
		h.current.Store(nil)
		return
	}
	old := h.current.Swap(r)
	if old == nil {
		return
	}
	h.drainsMu.Lock()
	h.drains = append(h.drains, &pendingClose[T]{
		r:        old,
		expireAt: time.Now().Add(h.drainGrace),
	})
	h.drainsMu.Unlock()
}

// Get is the concurrent-safe shim over the current reader's Get. Returns
// (zero, false) when the holder is empty. The RLock is held across the Load AND
// the read so a concurrent reaper can't munmap this reader mid-read.
func (h *SnapshotHolder[T]) Get(sku string) (T, bool) {
	h.readMu.RLock()
	defer h.readMu.RUnlock()
	r := h.current.Load()
	if r == nil {
		var zero T
		return zero, false
	}
	return r.Get(sku)
}

// Scan walks every (key, value) of the current reader in key-sorted order.
// Return false from fn to stop early. The RLock is held for the whole scan so
// the reader can't be munmapped mid-walk — keep fn cheap (collect then sort
// outside the callback for ranked output). No-op when the holder is empty.
func (h *SnapshotHolder[T]) Scan(fn func(key string, v T) bool) {
	h.readMu.RLock()
	defer h.readMu.RUnlock()
	r := h.current.Load()
	if r == nil {
		return
	}
	r.Scan(fn)
}

// Len returns the count of the current reader (0 if empty).
func (h *SnapshotHolder[T]) Len() int {
	h.readMu.RLock()
	defer h.readMu.RUnlock()
	r := h.current.Load()
	if r == nil {
		return 0
	}
	return r.Len()
}

// Close stops the reaper goroutine and closes the current reader + any
// pending drains without waiting for the grace period. Safe to call
// multiple times.
func (h *SnapshotHolder[T]) Close() error {
	var firstErr error
	h.once.Do(func() {
		close(h.stopCh)
		// Close under Lock so no in-flight Get is reading these readers' mmap
		// when we munmap them.
		h.readMu.Lock()
		r := h.current.Swap(nil)
		if r != nil {
			if err := r.Close(); err != nil {
				firstErr = err
			}
		}
		h.readMu.Unlock()
		h.drainsMu.Lock()
		drained := h.drains
		h.drains = nil
		h.drainsMu.Unlock()
		for _, d := range drained {
			h.readMu.Lock()
			err := d.r.Close()
			h.readMu.Unlock()
			if err != nil && firstErr == nil {
				firstErr = err
			}
		}
	})
	return firstErr
}

// DrainCount returns the number of readers currently in the drain queue.
// Useful for tests and operational metrics.
func (h *SnapshotHolder[T]) DrainCount() int {
	h.drainsMu.Lock()
	defer h.drainsMu.Unlock()
	return len(h.drains)
}

// reapNow scans the drain queue and Closes any readers whose grace has
// elapsed. Called by the sweeper goroutine on a tick.
func (h *SnapshotHolder[T]) reapNow() {
	now := time.Now()
	var alive []*pendingClose[T]
	var reaped []*pendingClose[T]
	h.drainsMu.Lock()
	for _, d := range h.drains {
		if !d.expireAt.After(now) {
			reaped = append(reaped, d)
		} else {
			alive = append(alive, d)
		}
	}
	h.drains = alive
	h.drainsMu.Unlock()
	for _, d := range reaped {
		// Lock briefly so the munmap can't race an in-flight Get on this reader.
		h.readMu.Lock()
		_ = d.r.Close()
		h.readMu.Unlock()
	}
}

func (h *SnapshotHolder[T]) sweepLoop() {
	interval := h.drainGrace / 4
	if interval < 10*time.Millisecond {
		interval = 10 * time.Millisecond
	}
	t := time.NewTicker(interval)
	defer t.Stop()
	for {
		select {
		case <-h.stopCh:
			return
		case <-t.C:
			h.reapNow()
		}
	}
}