Serving tier: scatter-gather fan-out, tail-latency, caching

go.mod

@@ -3,6 +3,7 @@ module openindex
 go 1.26
 
 require (
+	golang.org/x/sync v0.20.0
 	google.golang.org/grpc v1.81.1
 	google.golang.org/protobuf v1.36.11
 )

go.sum

@@ -24,6 +24,8 @@ go.opentelemetry.io/otel/trace v1.43.0 h1:BkNrHpup+4k4w+ZZ86CZoHHEkohws8AY+WTX09
 go.opentelemetry.io/otel/trace v1.43.0/go.mod h1:/QJhyVBUUswCphDVxq+8mld+AvhXZLhe+8WVFxiFff0=
 golang.org/x/net v0.51.0 h1:94R/GTO7mt3/4wIKpcR5gkGmRLOuE/2hNGeWq/GBIFo=
 golang.org/x/net v0.51.0/go.mod h1:aamm+2QF5ogm02fjy5Bb7CQ0WMt1/WVM7FtyaTLlA9Y=
+golang.org/x/sync v0.20.0 h1:e0PTpb7pjO8GAtTs2dQ6jYa5BWYlMuX047Dco/pItO4=
+golang.org/x/sync v0.20.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
 golang.org/x/sys v0.42.0 h1:omrd2nAlyT5ESRdCLYdm3+fMfNFE/+Rf4bDIQImRJeo=
 golang.org/x/sys v0.42.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
 golang.org/x/text v0.34.0 h1:oL/Qq0Kdaqxa1KbNeMKwQq0reLCCaFtqu2eNuSeNHbk=

serve/cache.go

@@ -0,0 +1,145 @@
+package serve
+
+import (
+	"container/list"
+	"context"
+	"sync"
+
+	"golang.org/x/sync/singleflight"
+
+	"openindex"
+)
+
+// Cache is the result-cache seam. The production in-process cache is Ristretto:
+// TinyLFU admission with sampled-LFU eviction, cost-based and concurrent, which
+// is best-in-class on hit ratio but deliberately drops some Set calls (a new
+// item may not be admitted), acceptable for a cache (doc 08.3). LRUCache below
+// is the in-process reference behind this seam.
+//
+// A cache key is the normalized query plus the snapshot id; the value is the
+// final ranked page. The front-end result cache serves a page with zero backend
+// work but only on an exact-query hit, so its ceiling is low (most unique
+// queries are singletons, doc 08.3); the posting-list cache carries more of the
+// load and lives in the leaf.
+type Cache interface {
+	Get(key string) ([]openindex.Result, bool)
+	Set(key string, val []openindex.Result)
+}
+
+// LRUCache is a fixed-capacity, least-recently-used result cache, safe for
+// concurrent use. It is the reference: a plain replacement policy that stands
+// in for Ristretto's admission policy so the serving path is testable without
+// the dependency. Unlike Ristretto it admits every Set, so a test sees a
+// deterministic hit pattern.
+type LRUCache struct {
+	mu       sync.Mutex
+	capacity int
+	ll       *list.List
+	items    map[string]*list.Element
+}
+
+type entry struct {
+	key string
+	val []openindex.Result
+}
+
+// NewLRUCache returns a cache holding at most capacity entries. A capacity <= 0
+// makes a cache that stores nothing, so every lookup misses.
+func NewLRUCache(capacity int) *LRUCache {
+	return &LRUCache{
+		capacity: capacity,
+		ll:       list.New(),
+		items:    make(map[string]*list.Element),
+	}
+}
+
+// Get returns the cached page for key and moves it to most-recently-used.
+func (c *LRUCache) Get(key string) ([]openindex.Result, bool) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	el, ok := c.items[key]
+	if !ok {
+		return nil, false
+	}
+	c.ll.MoveToFront(el)
+	return el.Value.(*entry).val, true
+}
+
+// Set stores val under key, evicting the least-recently-used entry if the cache
+// is over capacity.
+func (c *LRUCache) Set(key string, val []openindex.Result) {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	if c.capacity <= 0 {
+		return
+	}
+	if el, ok := c.items[key]; ok {
+		el.Value.(*entry).val = val
+		c.ll.MoveToFront(el)
+		return
+	}
+	el := c.ll.PushFront(&entry{key: key, val: val})
+	c.items[key] = el
+	if c.ll.Len() > c.capacity {
+		c.evict()
+	}
+}
+
+// Len reports the number of cached entries.
+func (c *LRUCache) Len() int {
+	c.mu.Lock()
+	defer c.mu.Unlock()
+	return c.ll.Len()
+}
+
+func (c *LRUCache) evict() {
+	el := c.ll.Back()
+	if el == nil {
+		return
+	}
+	c.ll.Remove(el)
+	delete(c.items, el.Value.(*entry).key)
+}
+
+// Loader fronts a Cache with single-flight stampede protection: when many
+// queries miss the same hot key at once, only one backend call runs and the
+// rest wait on its result (doc 08.3, doc 01.3). Without this, a popular query
+// expiring from the cache lets every concurrent request hit the backend at
+// once, which is exactly when the backend can least afford it.
+type Loader struct {
+	cache Cache
+	group singleflight.Group
+}
+
+// NewLoader wraps a cache.
+func NewLoader(cache Cache) *Loader {
+	return &Loader{cache: cache}
+}
+
+// Get returns the cached page for key, or runs load to compute it, stores it,
+// and returns it. Concurrent Gets for the same key while load is running share
+// the one in-flight call rather than each running load. The wait respects the
+// context: a caller whose deadline passes returns ctx.Err() even while the
+// shared load is still running for the others. A load error is not cached.
+func (l *Loader) Get(ctx context.Context, key string, load func(context.Context) ([]openindex.Result, error)) ([]openindex.Result, error) {
+	if v, ok := l.cache.Get(key); ok {
+		return v, nil
+	}
+	ch := l.group.DoChan(key, func() (any, error) {
+		v, err := load(ctx)
+		if err != nil {
+			return nil, err
+		}
+		l.cache.Set(key, v)
+		return v, nil
+	})
+	select {
+	case <-ctx.Done():
+		return nil, ctx.Err()
+	case res := <-ch:
+		if res.Err != nil {
+			return nil, res.Err
+		}
+		return res.Val.([]openindex.Result), nil
+	}
+}

serve/cache_test.go

@@ -0,0 +1,144 @@
+package serve
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"openindex"
+)
+
+func TestLRUGetSet(t *testing.T) {
+	c := NewLRUCache(2)
+	if _, ok := c.Get("missing"); ok {
+		t.Fatal("empty cache should miss")
+	}
+	c.Set("a", []openindex.Result{res(1, 9)})
+	got, ok := c.Get("a")
+	if !ok || len(got) != 1 || got[0].Doc.Local != 1 {
+		t.Fatalf("get after set failed: %v %v", got, ok)
+	}
+}
+
+func TestLRUEviction(t *testing.T) {
+	c := NewLRUCache(2)
+	c.Set("a", []openindex.Result{res(1, 1)})
+	c.Set("b", []openindex.Result{res(2, 2)})
+	_, _ = c.Get("a")                         // a is now most-recently-used
+	c.Set("c", []openindex.Result{res(3, 3)}) // evicts b, the least recent
+	if _, ok := c.Get("b"); ok {
+		t.Fatal("b should have been evicted")
+	}
+	if _, ok := c.Get("a"); !ok {
+		t.Fatal("a was used recently and should survive")
+	}
+	if c.Len() != 2 {
+		t.Fatalf("cache should hold its capacity, got %d", c.Len())
+	}
+}
+
+func TestLRUUpdateInPlace(t *testing.T) {
+	c := NewLRUCache(2)
+	c.Set("a", []openindex.Result{res(1, 1)})
+	c.Set("a", []openindex.Result{res(1, 5)})
+	got, _ := c.Get("a")
+	if got[0].Score != 5 {
+		t.Fatalf("set should overwrite, got score %g", got[0].Score)
+	}
+	if c.Len() != 1 {
+		t.Fatalf("overwrite should not grow the cache, got %d", c.Len())
+	}
+}
+
+func TestLRUZeroCapacityStoresNothing(t *testing.T) {
+	c := NewLRUCache(0)
+	c.Set("a", []openindex.Result{res(1, 1)})
+	if _, ok := c.Get("a"); ok {
+		t.Fatal("a zero-capacity cache should store nothing")
+	}
+}
+
+func TestLoaderCollapsesStampede(t *testing.T) {
+	l := NewLoader(NewLRUCache(16))
+	var loads int32
+	const n = 50
+	var start, done sync.WaitGroup
+	start.Add(1)
+	done.Add(n)
+	got := make([][]openindex.Result, n)
+	for i := range n {
+		go func() {
+			defer done.Done()
+			start.Wait() // release all at once so they collide on the miss
+			r, err := l.Get(context.Background(), "hot", func(context.Context) ([]openindex.Result, error) {
+				atomic.AddInt32(&loads, 1)
+				time.Sleep(40 * time.Millisecond) // hold the flight open
+				return []openindex.Result{res(1, 9)}, nil
+			})
+			if err != nil {
+				t.Errorf("get %d: %v", i, err)
+			}
+			got[i] = r
+		}()
+	}
+	start.Done()
+	done.Wait()
+	if n := atomic.LoadInt32(&loads); n != 1 {
+		t.Fatalf("stampede should collapse to one backend call, got %d", n)
+	}
+	for i := range got {
+		if len(got[i]) != 1 || got[i][0].Doc.Local != 1 {
+			t.Fatalf("caller %d got the wrong result: %v", i, got[i])
+		}
+	}
+}
+
+func TestLoaderServesFromCache(t *testing.T) {
+	l := NewLoader(NewLRUCache(16))
+	var loads int32
+	load := func(context.Context) ([]openindex.Result, error) {
+		atomic.AddInt32(&loads, 1)
+		return []openindex.Result{res(1, 9)}, nil
+	}
+	for range 3 {
+		if _, err := l.Get(context.Background(), "k", load); err != nil {
+			t.Fatal(err)
+		}
+	}
+	if atomic.LoadInt32(&loads) != 1 {
+		t.Fatalf("after the first load the rest should hit the cache, got %d loads", loads)
+	}
+}
+
+func TestLoaderDoesNotCacheErrors(t *testing.T) {
+	l := NewLoader(NewLRUCache(16))
+	boom := errors.New("backend down")
+	if _, err := l.Get(context.Background(), "k", func(context.Context) ([]openindex.Result, error) {
+		return nil, boom
+	}); !errors.Is(err, boom) {
+		t.Fatalf("first call should surface the error, got %v", err)
+	}
+	// A second call must re-run load, since the error was not cached.
+	got, err := l.Get(context.Background(), "k", func(context.Context) ([]openindex.Result, error) {
+		return []openindex.Result{res(2, 1)}, nil
+	})
+	if err != nil || len(got) != 1 || got[0].Doc.Local != 2 {
+		t.Fatalf("second call should recompute: %v %v", got, err)
+	}
+}
+
+func TestLoaderRespectsContext(t *testing.T) {
+	l := NewLoader(NewLRUCache(16))
+	ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
+	defer cancel()
+	_, err := l.Get(ctx, "k", func(c context.Context) ([]openindex.Result, error) {
+		<-c.Done()
+		return nil, c.Err()
+	})
+	if !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("a caller whose deadline passes should get the deadline error, got %v", err)
+	}
+}

serve/coordinator.go

@@ -0,0 +1,62 @@
+package serve
+
+import (
+	"context"
+	"errors"
+	"math"
+
+	"openindex"
+)
+
+// ErrInsufficientShards is returned when too few shards answered for the result
+// to be trustworthy. It is the floor under the good-enough cutoff: dropping a
+// slow shard is fine, but a page built from a small minority of shards is
+// missing too much to serve.
+var ErrInsufficientShards = errors.New("serve: too few shards responded")
+
+// Coordinator is a serving-tree node that fans a query to its children and
+// merges their replies. The same type is the root over aggregators and an
+// aggregator over leaves; only the children differ. It applies the good-enough
+// cutoff from doc 08.2, which is not a separate timer but the combination of
+// the per-shard sub-deadline (slow shards simply do not answer in time) and a
+// minimum-responded floor checked after the gather.
+type Coordinator struct {
+	children []Leaf
+	cfg      FanoutConfig
+	// minResponded is the fraction of children that must answer for the result
+	// to be served, in (0,1]. A value of 0.95 returns once 95 percent have
+	// replied and treats the rest as the tail to drop.
+	minResponded float64
+}
+
+// NewCoordinator builds a node over the given children. minResponded is clamped
+// to (0,1]; a zero or negative value selects 1.0 (every child must answer),
+// which is the strict default a caller relaxes deliberately.
+func NewCoordinator(children []Leaf, cfg FanoutConfig, minResponded float64) *Coordinator {
+	if minResponded <= 0 || minResponded > 1 {
+		minResponded = 1
+	}
+	return &Coordinator{children: children, cfg: cfg, minResponded: minResponded}
+}
+
+// Search fans req out to the children, applies the good-enough cutoff, and
+// returns the merged global top-k. It returns ErrInsufficientShards when fewer
+// than the required fraction of children answered, so a degraded page is never
+// silently served as if it were complete.
+func (c *Coordinator) Search(ctx context.Context, req Request) ([]openindex.Result, error) {
+	if len(c.children) == 0 {
+		return nil, nil
+	}
+	g := Scatter(ctx, c.children, req, c.cfg)
+
+	need := max(int(math.Ceil(c.minResponded*float64(len(c.children)))), 1)
+	if g.Responded() < need {
+		// Surface the context error if that is why shards were lost, since a
+		// cancelled query is a different failure from a flaky fleet.
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
+		return nil, ErrInsufficientShards
+	}
+	return MergeTopK(g.OKResponses(), req.K), nil
+}