feat: add opt-in V3 delta search refresh

api/v1-audit.txt

@@ -288,6 +288,9 @@ field VersionedOptions.RefreshDebounce time.Duration	ok	versioned.go:58 additive
 field VersionedStatus.CompletedBuilds uint64	ok	versioned.go:73 additive R2/R3 contract; source-span and limit parity verified in scan_test.go:1, cache and coalescing in versioned_reuse_test.go:1
 field VersionedStatus.DownloadedBuckets int	ok	versioned.go:73 additive R2/R3 contract; source-span and limit parity verified in scan_test.go:1, cache and coalescing in versioned_reuse_test.go:1
 field VersionedStatus.ReusedBuckets int	ok	versioned.go:73 additive R2/R3 contract; source-span and limit parity verified in scan_test.go:1, cache and coalescing in versioned_reuse_test.go:1
+field VersionedOptions.DeltaSearch bool	ok	versioned.go:60 opt-in small-update search; publication and compaction verified in versioned_overlay_test.go:1
+field VersionedStatus.DeltaKeywords int	ok	versioned.go:80 overlay keyword count verified in versioned_overlay_test.go:1
+field VersionedStatus.DeltaSearch bool	ok	versioned.go:80 overlay state verified in versioned_overlay_test.go:1
 method (*AhoCorasick) MaskText(ctx context.Context, text string, mask rune, opts *RewriteOptions) (*RewriteResult, error)	ok	rewrite.go:41 additive R2/R3 contract; source-span and limit parity verified in scan_test.go:1, cache and coalescing in versioned_reuse_test.go:1
 method (*AhoCorasick) ReplaceText(ctx context.Context, text, replacement string, opts *RewriteOptions) (*RewriteResult, error)	ok	rewrite.go:35 additive R2/R3 contract; source-span and limit parity verified in scan_test.go:1, cache and coalescing in versioned_reuse_test.go:1
 method (*AhoCorasick) Scan(ctx context.Context, text string, opts *ScanOptions) (*ScanResult, error)	ok	scan.go:70 additive R2/R3 contract; source-span and limit parity verified in scan_test.go:1, cache and coalescing in versioned_reuse_test.go:1

api/v1.txt

@@ -129,6 +129,7 @@ field V2CopyResult.Count int
 field V2CopyResult.SourceVersion string
 field V2CopyResult.Write *WriteResult
 field VersionedOptions.CaseSensitive bool
+field VersionedOptions.DeltaSearch bool
 field VersionedOptions.LeaseDuration time.Duration
 field VersionedOptions.LeaseRefresh time.Duration
 field VersionedOptions.PollInterval time.Duration
@@ -140,6 +141,8 @@ field VersionedStatus.BuildDuration time.Duration
 field VersionedStatus.BuildStarted time.Time
 field VersionedStatus.Building bool
 field VersionedStatus.CompletedBuilds uint64
+field VersionedStatus.DeltaKeywords int
+field VersionedStatus.DeltaSearch bool
 field VersionedStatus.DownloadedBuckets int
 field VersionedStatus.LastError string
 field VersionedStatus.ReusedBuckets int

changes/unreleased/20260913-v3-delta-search.yaml

@@ -0,0 +1,5 @@
+kind: Added
+body: 'V3 collections can opt into small-update search with `DeltaSearch`, publishing additions and deletions before a deferred full rebuild.'
+time: 2026-09-13T21:00:00+09:00
+custom:
+  Issue: "255"

docs/content/reference/versioned.md

@@ -82,6 +82,17 @@ A successful write means the Redis commit landed. The calling instance and every
 instance may still be searching the previous engine. `WaitForVersion` waits for that
 commit or a later one and honors cancellation. `Status` exposes the observed active
 version, the serving version, build start and duration, and the most recent error.
+`DeltaSearch` and `DeltaKeywords` report whether the serving version uses a small
+change set and how many additions and deletions it contains.
+
+Set `VersionedOptions.DeltaSearch` to enable the experimental small-update path for
+`MemoryEfficient`. Changes of at most 128 keywords download only changed buckets and
+publish a base automaton plus an additions automaton and deletion set. `WaitForVersion`
+completes when this search view is ready. After one second without another change, a
+background build replaces it with a full automaton; later changes or build failures
+leave the published search view intact. Larger changes and other presets continue to
+use the full rebuild path. This option remains off by default pending million-keyword
+Redis and Valkey latency, search p95, and peak RSS validation.
 
 | Refresh behavior | |
 |---|---|

internal/engine/overlay.go

@@ -0,0 +1,226 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package engine
+
+import (
+	"slices"
+	"unicode/utf8"
+)
+
+// Overlay combines an immutable base with a small replacement set. Its inputs
+// must not be mutated after construction. It is used only by V3 snapshots.
+func Overlay(base, added *Engine, removed map[string]struct{}) *Engine {
+	i := &overlayEngine{base: base, added: added, removed: removed}
+	return &Engine{preset: base.preset, impl: i, maxKeywordRunes: max(base.MaxKeywordRunes(), added.MaxKeywordRunes())}
+}
+
+type overlayEngine struct {
+	base, added *Engine
+	removed     map[string]struct{}
+}
+
+type overlayMatch struct {
+	keyword string
+	start   int
+	end     int
+	source  int
+	order   int
+}
+
+func (o *overlayEngine) buildFromKeywords(map[string]struct{}) { panic("engine: immutable overlay") }
+
+func (o *overlayEngine) matchString(text string, emit func(string, int, int) bool) {
+	i := 0
+	o.matchStream(func() (rune, bool) {
+		if i == len(text) {
+			return 0, false
+		}
+		r, size := utf8.DecodeRuneInString(text[i:])
+		i += size
+		return r, true
+	}, emit)
+}
+
+func (o *overlayEngine) matchStream(next func() (rune, bool), emit func(string, int, int) bool) {
+	base, baseOK := o.base.impl.(*memEfficientEngine)
+	added, addedOK := o.added.impl.(*memEfficientEngine)
+	if !baseOK || !addedOK {
+		o.matchStreamWindow(next, emit)
+		return
+	}
+	baseState, addedState, end := 0, 0, 0
+	matches := make([]overlayMatch, 0, 8)
+	for {
+		ch, ok := next()
+		if !ok {
+			return
+		}
+		end++
+		matches = matches[:0]
+		baseState = base.advance(baseState, ch)
+		base.trie.out.emitChain(baseState, end, func(k string, start, end int) bool {
+			if _, removed := o.removed[k]; !removed {
+				matches = append(matches, overlayMatch{k, start, end, 0, len(matches)})
+			}
+			return true
+		})
+		addedState = added.advance(addedState, ch)
+		added.trie.out.emitChain(addedState, end, func(k string, start, end int) bool {
+			matches = append(matches, overlayMatch{k, start, end, 1, len(matches)})
+			return true
+		})
+		slices.SortStableFunc(matches, func(a, b overlayMatch) int {
+			if a.start != b.start {
+				return a.start - b.start
+			}
+			if a.source != b.source {
+				return a.source - b.source
+			}
+			return a.order - b.order
+		})
+		for _, m := range matches {
+			if !emit(m.keyword, m.start, m.end) {
+				return
+			}
+		}
+	}
+}
+
+// Other presets can still be overlaid. Keep only enough input to decide the
+// matches ending at the current rune, so a stopped callback never reads ahead.
+func (o *overlayEngine) matchStreamWindow(next func() (rune, bool), emit func(string, int, int) bool) {
+	longest := max(o.base.MaxKeywordRunes(), o.added.MaxKeywordRunes())
+	window := make([]rune, 0, longest)
+	end := 0
+	for {
+		ch, ok := next()
+		if !ok {
+			return
+		}
+		end++
+		window = append(window, ch)
+		if len(window) > longest {
+			copy(window, window[1:])
+			window = window[:longest]
+		}
+		startOffset := end - len(window)
+		matches := make([]overlayMatch, 0, 8)
+		o.base.MatchString(string(window), func(k string, start, localEnd int) bool {
+			if localEnd == len(window) {
+				if _, removed := o.removed[k]; !removed {
+					matches = append(matches, overlayMatch{k, startOffset + start, end, 0, len(matches)})
+				}
+			}
+			return true
+		})
+		o.added.MatchString(string(window), func(k string, start, localEnd int) bool {
+			if localEnd == len(window) {
+				matches = append(matches, overlayMatch{k, startOffset + start, end, 1, len(matches)})
+			}
+			return true
+		})
+		slices.SortStableFunc(matches, func(a, b overlayMatch) int {
+			if a.start != b.start {
+				return a.start - b.start
+			}
+			if a.source != b.source {
+				return a.source - b.source
+			}
+			return a.order - b.order
+		})
+		for _, m := range matches {
+			if !emit(m.keyword, m.start, m.end) {
+				return
+			}
+		}
+	}
+}
+
+func (e *memEfficientEngine) advance(state int, ch rune) int {
+	if e.bloom.skipAtRoot(state == 0, ch) {
+		return 0
+	}
+	for {
+		if next, ok := e.trie.nodes[state].next(ch); ok {
+			return next
+		}
+		if state == 0 {
+			return 0
+		}
+		state = e.trie.nodes[state].fail
+	}
+}
+
+func (o *overlayEngine) find(text string) []string {
+	if !o.added.Contains(text) {
+		base := o.base.Find(text)
+		out := base[:0]
+		for _, k := range base {
+			if _, deleted := o.removed[k]; !deleted {
+				out = append(out, k)
+			}
+		}
+		return out
+	}
+	out := []string{}
+	o.matchString(text, func(k string, _, _ int) bool { out = append(out, k); return true })
+	return out
+}
+func (o *overlayEngine) findSet(text string) []string {
+	if !o.added.Contains(text) {
+		base := o.base.FindSet(text)
+		out := base[:0]
+		for _, k := range base {
+			if _, deleted := o.removed[k]; !deleted {
+				out = append(out, k)
+			}
+		}
+		return out
+	}
+	out := []string{}
+	seen := make(map[string]struct{})
+	o.matchString(text, func(k string, _, _ int) bool {
+		if _, ok := seen[k]; !ok {
+			seen[k] = struct{}{}
+			out = append(out, k)
+		}
+		return true
+	})
+	return out
+}
+func (o *overlayEngine) findIndex(text string) map[string][]int {
+	if !o.added.Contains(text) {
+		out := o.base.FindIndex(text)
+		for k := range o.removed {
+			delete(out, k)
+		}
+		return out
+	}
+	out := make(map[string][]int)
+	o.matchString(text, func(k string, start, _ int) bool { out[k] = append(out[k], start); return true })
+	return out
+}
+func (o *overlayEngine) contains(text string) bool {
+	if o.added.Contains(text) {
+		return true
+	}
+	if len(o.removed) == 0 {
+		return o.base.Contains(text)
+	}
+	found := false
+	o.base.MatchString(text, func(k string, _, _ int) bool {
+		if _, deleted := o.removed[k]; !deleted {
+			found = true
+			return false
+		}
+		return true
+	})
+	return found
+}
+func (o *overlayEngine) info() *InMemoryInfo {
+	i := *o.base.Info()
+	i.Keywords += o.added.Info().Keywords - len(o.removed)
+	i.MemoryBytes += o.added.Info().MemoryBytes + int64(len(o.removed))*32
+	i.TrieDepth = max(i.TrieDepth, o.added.Info().TrieDepth)
+	return &i
+}

internal/engine/overlay_test.go

@@ -0,0 +1,75 @@
+// SPDX-License-Identifier: Apache-2.0
+
+package engine
+
+import (
+	"reflect"
+	"strings"
+	"testing"
+)
+
+func TestOverlayMatchesRebuild(t *testing.T) {
+	base := New(PresetMemoryEfficient)
+	base.Build(map[string]struct{}{"he": {}, "she": {}, "hers": {}, "한국": {}, "old": {}})
+	added := New(PresetMemoryEfficient)
+	added.Build(map[string]struct{}{"her": {}, "한국어": {}})
+	overlay := Overlay(base, added, map[string]struct{}{"old": {}})
+	full := New(PresetMemoryEfficient)
+	full.Build(map[string]struct{}{"he": {}, "she": {}, "hers": {}, "한국": {}, "her": {}, "한국어": {}})
+	text := strings.Repeat("ushers old 한국어 ", 300)
+	if got, want := overlay.Find(text), full.Find(text); !reflect.DeepEqual(got, want) {
+		t.Fatal("find differs")
+	}
+	if got, want := overlay.FindIndex(text), full.FindIndex(text); !reflect.DeepEqual(got, want) {
+		t.Fatal("index differs")
+	}
+	if got, want := overlay.FindSet(text), full.FindSet(text); !reflect.DeepEqual(got, want) {
+		t.Fatal("set differs")
+	}
+	collect := func(e *Engine) []string {
+		var out []string
+		runes := []rune(text)
+		i := 0
+		e.Stream(func() (rune, bool) {
+			if i == len(runes) {
+				return 0, false
+			}
+			r := runes[i]
+			i++
+			return r, true
+		},
+			func(k string, start, end int) bool { out = append(out, k); return true })
+		return out
+	}
+	if got, want := collect(overlay), collect(full); !reflect.DeepEqual(got, want) {
+		t.Fatal("stream differs")
+	}
+}
+
+func TestOverlayStopsAtFirstMatch(t *testing.T) {
+	base := New(PresetMemoryEfficient)
+	base.Build(map[string]struct{}{"a": {}, "aa": {}})
+	added := New(PresetMemoryEfficient)
+	added.Build(map[string]struct{}{"aaa": {}})
+	overlay := Overlay(base, added, nil)
+	text := strings.Repeat("a", 100000)
+	called := 0
+	overlay.MatchString(text, func(k string, start, end int) bool {
+		called++
+		return false
+	})
+	if called != 1 {
+		t.Fatalf("callback called %d times", called)
+	}
+	read := 0
+	overlay.Stream(func() (rune, bool) {
+		read++
+		if read > 1 {
+			t.Fatal("stream read beyond first match")
+		}
+		return 'a', true
+	}, func(string, int, int) bool { return false })
+	if read != 1 {
+		t.Fatalf("stream read %d runes", read)
+	}
+}