From 0f2312e1ed0c2290536360029725001a62e2cdca Mon Sep 17 00:00:00 2001 From: Sung-Kyu Yoo Date: Sun, 13 Sep 2026 22:25:21 +0900 Subject: [PATCH 1/4] feat: add opt-in V3 delta search refresh --- api/v1-audit.txt | 3 + api/v1.txt | 3 + .../unreleased/20260913-v3-delta-search.yaml | 3 + docs/content/reference/versioned.md | 11 ++ internal/engine/overlay.go | 182 ++++++++++++++++++ internal/engine/overlay_test.go | 47 +++++ pkg/acor/versioned.go | 33 +++- pkg/acor/versioned_overlay.go | 177 +++++++++++++++++ pkg/acor/versioned_overlay_test.go | 80 ++++++++ pkg/acor/versioned_scale_test.go | 36 ++-- 10 files changed, 556 insertions(+), 19 deletions(-) create mode 100644 changes/unreleased/20260913-v3-delta-search.yaml create mode 100644 internal/engine/overlay.go create mode 100644 internal/engine/overlay_test.go create mode 100644 pkg/acor/versioned_overlay.go create mode 100644 pkg/acor/versioned_overlay_test.go diff --git a/api/v1-audit.txt b/api/v1-audit.txt index eee8291..57a6a53 100644 --- a/api/v1-audit.txt +++ b/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 diff --git a/api/v1.txt b/api/v1.txt index 7ef9a34..2bb5b69 100644 --- a/api/v1.txt +++ b/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 diff --git a/changes/unreleased/20260913-v3-delta-search.yaml b/changes/unreleased/20260913-v3-delta-search.yaml new file mode 100644 index 0000000..7b7d76c --- /dev/null +++ b/changes/unreleased/20260913-v3-delta-search.yaml @@ -0,0 +1,3 @@ +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 diff --git a/docs/content/reference/versioned.md b/docs/content/reference/versioned.md index e2de4a1..c4cb6f2 100644 --- a/docs/content/reference/versioned.md +++ b/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 | | |---|---| diff --git a/internal/engine/overlay.go b/internal/engine/overlay.go new file mode 100644 index 0000000..ebebc36 --- /dev/null +++ b/internal/engine/overlay.go @@ -0,0 +1,182 @@ +// SPDX-License-Identifier: Apache-2.0 + +package engine + +import ( + "slices" +) + +// 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) { + var matches []overlayMatch + order := 0 + o.base.MatchString(text, func(k string, start, end int) bool { + if _, deleted := o.removed[k]; !deleted { + matches = append(matches, overlayMatch{k, start, end, 0, order}) + } + order++ + return true + }) + order = 0 + o.added.MatchString(text, func(k string, start, end int) bool { + matches = append(matches, overlayMatch{k, start, end, 1, order}) + order++ + return true + }) + slices.SortStableFunc(matches, func(a, b overlayMatch) int { + if a.end != b.end { + return a.end - b.end + } + 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 (o *overlayEngine) matchStream(next func() (rune, bool), emit func(string, int, int) bool) { + const chunk = 4096 + longest := max(o.base.MaxKeywordRunes(), o.added.MaxKeywordRunes()) + window := make([]rune, 0, chunk+longest) + windowStart := 0 + total := 0 + stopped := false + emittedEnd := 0 + process := func() { + o.matchString(string(window), func(k string, from, end int) bool { + if windowStart+end <= emittedEnd { + return true + } + if !emit(k, windowStart+from, windowStart+end) { + stopped = true + return false + } + return true + }) + emittedEnd = total + keep := min(len(window), max(0, longest-1)) + windowStart = total - keep + copy(window, window[len(window)-keep:]) + window = window[:keep] + } + for { + r, ok := next() + if !ok { + break + } + window = append(window, r) + total++ + if len(window) >= chunk+longest-1 { + process() + if stopped { + return + } + } + } + if len(window) > 0 && !stopped { + process() + } +} + +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 +} diff --git a/internal/engine/overlay_test.go b/internal/engine/overlay_test.go new file mode 100644 index 0000000..bd39a32 --- /dev/null +++ b/internal/engine/overlay_test.go @@ -0,0 +1,47 @@ +// 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") + } +} diff --git a/pkg/acor/versioned.go b/pkg/acor/versioned.go index 67b0865..c7eb0d9 100644 --- a/pkg/acor/versioned.go +++ b/pkg/acor/versioned.go @@ -34,6 +34,9 @@ const ( v3MillisPerSecond = 1000 v3PageHint = 1024 v3DebounceDefault = 20 * time.Millisecond + v3OverlayLimit = 128 + v3CompactDelay = time.Second + v3CompactPoll = 100 * time.Millisecond ) var ( @@ -59,6 +62,9 @@ type VersionedOptions struct { Redis AhoCorasickArgs CaseSensitive bool Preset Preset + // DeltaSearch enables the small-update search path for MemoryEfficient. + // It remains opt-in until the release-scale latency and RSS gates are met. + DeltaSearch bool // PollInterval defaults to 30 seconds; negative values are invalid. PollInterval time.Duration // RefreshDebounce coalesces bursts before a build, default 20ms. Negative values are invalid. @@ -82,6 +88,8 @@ type VersionedStatus struct { ReusedBuckets int // CompletedBuilds counts successful engines installed by this instance. CompletedBuilds uint64 + DeltaSearch bool + DeltaKeywords int } // WriteResult describes an atomic commit. OperationID remains available on an @@ -106,11 +114,17 @@ type v3Bucket struct { Checksum string } type v3Engine struct { - engine *matchengine.Engine - version Version - sequence uint64 - manifest *v3Manifest - buckets *[v3BucketCount][]string + engine *matchengine.Engine + base *matchengine.Engine + baseBuckets *[v3BucketCount][]string + baseManifest *v3Manifest + added map[string]struct{} + removed map[string]struct{} + version Version + sequence uint64 + manifest *v3Manifest + buckets *[v3BucketCount][]string + installed time.Time } // VersionedCollection owns a V3 dictionary and a background engine refresher. @@ -183,6 +197,8 @@ func OpenVersioned(ctx context.Context, opts *VersionedOptions) (*VersionedColle } v.wg.Add(1) go v.refreshLoop() + v.wg.Add(1) + go v.compactLoop() return v, nil } @@ -270,6 +286,9 @@ func (v *VersionedCollection) refresh(ctx context.Context) error { if err != nil { return err } + if used, err := v.tryOverlay(ctx, s, buckets, downloaded, reused); used || err != nil { + return err + } e := matchengine.New(enginePreset(v.opts.Preset)) if buildErr := e.BuildSequenceContext(ctx, bucketSequence(buckets), s.Count()); buildErr != nil { return buildErr @@ -284,13 +303,15 @@ func (v *VersionedCollection) refresh(ctx context.Context) error { return nil } func (v *VersionedCollection) installEngine(s *Snapshot, e *matchengine.Engine, buckets *[v3BucketCount][]string, downloaded, reused int) { - v.current.Store(&v3Engine{engine: e, version: s.Version(), sequence: s.manifest.Sequence, manifest: s.manifest, buckets: buckets}) + v.current.Store(&v3Engine{engine: e, version: s.Version(), sequence: s.manifest.Sequence, manifest: s.manifest, buckets: buckets, installed: time.Now()}) v.mu.Lock() v.status.ServingVersion = s.Version() v.status.LastError = "" v.status.DownloadedBuckets = downloaded v.status.ReusedBuckets = reused v.status.CompletedBuilds++ + v.status.DeltaSearch = false + v.status.DeltaKeywords = 0 v.mu.Unlock() } func (v *VersionedCollection) refreshLoop() { diff --git a/pkg/acor/versioned_overlay.go b/pkg/acor/versioned_overlay.go new file mode 100644 index 0000000..5057f0d --- /dev/null +++ b/pkg/acor/versioned_overlay.go @@ -0,0 +1,177 @@ +// SPDX-License-Identifier: Apache-2.0 + +package acor + +import ( + "context" + "iter" + "time" + + matchengine "github.com/skyoo2003/acor/internal/engine" +) + +func v3MapSequence(words map[string]struct{}) iter.Seq[string] { + return func(yield func(string) bool) { + for word := range words { + if !yield(word) { + return + } + } + } +} + +// Compare the target to the dictionary represented by the base automaton. +// Unchanged buckets share their backing slice and cost no per-word work. +func v3BucketDelta(previous *v3Engine, target *v3Manifest, next *[v3BucketCount][]string) (added, removed map[string]struct{}) { + base := previous.baseBuckets + if base == nil { + base = previous.buckets + } + manifest := previous.baseManifest + if manifest == nil { + manifest = previous.manifest + } + added = make(map[string]struct{}) + removed = make(map[string]struct{}) + for i := range next { + if sameBucket(manifest.Buckets[i], target.Buckets[i]) { + continue + } + old := make(map[string]struct{}, len(base[i])) + for _, word := range base[i] { + old[word] = struct{}{} + } + for _, word := range next[i] { + if _, ok := old[word]; ok { + delete(old, word) + } else { + added[word] = struct{}{} + } + } + for word := range old { + removed[word] = struct{}{} + } + } + return added, removed +} + +func (v *VersionedCollection) tryOverlay(ctx context.Context, s *Snapshot, buckets *[v3BucketCount][]string, + downloaded, reused int) (bool, error) { + previous := v.current.Load() + if previous == nil || !v.opts.DeltaSearch || v.opts.Preset != PresetMemoryEfficient { + return false, nil + } + base := previous.base + if base == nil { + base = previous.engine + } + added, removed := v3BucketDelta(previous, s.manifest, buckets) + if len(added)+len(removed) > v3OverlayLimit { + return false, nil + } + addition := matchengine.New(enginePreset(v.opts.Preset)) + if err := addition.BuildSequenceContext(ctx, v3MapSequence(added), len(added)); err != nil { + return true, err + } + if err := s.lease.check(ctx); err != nil { + return true, err + } + if err := v.check(ctx); err != nil { + return true, err + } + v.installOverlay(s, base, addition, added, removed, buckets, downloaded, reused) + return true, nil +} + +func (v *VersionedCollection) installOverlay(s *Snapshot, base, addition *matchengine.Engine, + added, removed map[string]struct{}, buckets *[v3BucketCount][]string, downloaded, reused int) { + previous := v.current.Load() + baseBuckets := previous.baseBuckets + if baseBuckets == nil { + baseBuckets = previous.buckets + } + baseManifest := previous.baseManifest + if baseManifest == nil { + baseManifest = previous.manifest + } + v.current.Store(&v3Engine{engine: matchengine.Overlay(base, addition, removed), base: base, baseBuckets: baseBuckets, + baseManifest: baseManifest, + added: added, removed: removed, version: s.Version(), sequence: s.manifest.Sequence, + manifest: s.manifest, buckets: buckets, installed: time.Now()}) + v.mu.Lock() + v.status.ServingVersion = s.Version() + v.status.LastError = "" + v.status.DownloadedBuckets = downloaded + v.status.ReusedBuckets = reused + v.status.CompletedBuilds++ + v.status.DeltaSearch = true + v.status.DeltaKeywords = len(added) + len(removed) + v.mu.Unlock() +} + +func (v *VersionedCollection) compactLoop() { + defer v.wg.Done() + ticker := time.NewTicker(v3CompactPoll) + defer ticker.Stop() + for { + select { + case <-v.ctx.Done(): + return + case <-ticker.C: + v.compactIfIdle(v.ctx) + } + } +} + +func (v *VersionedCollection) compactIfIdle(ctx context.Context) { + current := v.current.Load() + if current == nil || current.base == nil || time.Since(current.installed) < v3CompactDelay { + return + } + v.refreshMu.Lock() + defer v.refreshMu.Unlock() + if v.current.Load() != current || ctx.Err() != nil { + return + } + start := time.Now() + v.mu.Lock() + v.status.Building = true + v.status.BuildStarted = start + v.mu.Unlock() + defer func() { + v.mu.Lock() + v.status.Building = false + v.status.BuildDuration = time.Since(start) + v.mu.Unlock() + }() + e := matchengine.New(enginePreset(v.opts.Preset)) + if err := e.BuildSequenceContext(ctx, bucketSequence(current.buckets), current.manifest.Count); err != nil { + v.recordCompactError(err) + return + } + active, err := v.client.Get(ctx, v.key("active")).Result() + if err != nil { + v.recordCompactError(err) + return + } + if Version(active) != current.version || v.current.Load() != current || ctx.Err() != nil { + return + } + v.current.Store(&v3Engine{engine: e, version: current.version, sequence: current.sequence, + manifest: current.manifest, buckets: current.buckets, installed: time.Now()}) + v.mu.Lock() + v.status.LastError = "" + v.status.CompletedBuilds++ + v.status.DeltaSearch = false + v.status.DeltaKeywords = 0 + v.mu.Unlock() +} + +func (v *VersionedCollection) recordCompactError(err error) { + if v.ctx.Err() != nil { + return + } + v.mu.Lock() + v.status.LastError = err.Error() + v.mu.Unlock() +} diff --git a/pkg/acor/versioned_overlay_test.go b/pkg/acor/versioned_overlay_test.go new file mode 100644 index 0000000..8d7ee5d --- /dev/null +++ b/pkg/acor/versioned_overlay_test.go @@ -0,0 +1,80 @@ +// SPDX-License-Identifier: Apache-2.0 + +package acor + +import ( + "context" + "fmt" + "reflect" + "strings" + "testing" + "time" + + "github.com/alicebob/miniredis/v2" +) + +//nolint:gocyclo // Lifecycle assertions cover publication, search, and compaction together. +func TestVersionedOverlayLifecycle(t *testing.T) { + ctx := context.Background() + server := miniredis.RunT(t) + v, err := OpenVersioned(ctx, &VersionedOptions{ + Redis: AhoCorasickArgs{Addr: server.Addr(), Name: "overlay"}, + DeltaSearch: true, PollInterval: 20 * time.Millisecond, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = v.Close() }) + base := make([]string, 129) + for i := range base { + base[i] = fmt.Sprintf("word-%03d", i) + } + r, err := v.Replace(ctx, v.Status().ServingVersion, base) + if err != nil { + t.Fatal(err) + } + waitV3(t, v, r.Version) + if v.Status().DeltaSearch { + t.Fatal("large replacement used overlay") + } + r, err = v.AddMany(ctx, r.Version, []string{"she", "한국어"}) + if err != nil { + t.Fatal(err) + } + waitV3(t, v, r.Version) + if s := v.Status(); !s.DeltaSearch || s.DeltaKeywords != 2 { + t.Fatalf("status=%+v", s) + } + r, err = v.Remove(ctx, r.Version, "word-000") + if err != nil { + t.Fatal(err) + } + waitV3(t, v, r.Version) + if s := v.Status(); !s.DeltaSearch || s.DeltaKeywords != 3 { + t.Fatalf("status=%+v", s) + } + text := "SHE 한국어 word-000 word-001" + got, err := v.Find(ctx, text) + if err != nil || !reflect.DeepEqual(got, []string{"she", "한국어", "word-001"}) { + t.Fatal(got, err) + } + var stream []Match + if err = v.FindStream(ctx, strings.NewReader(strings.Repeat(text+" ", 300)), func(m Match) bool { stream = append(stream, m); return true }); err != nil { + t.Fatal(err) + } + all, err := v.FindMatches(ctx, strings.Repeat(text+" ", 300), nil) + if err != nil || !reflect.DeepEqual(stream, all) { + t.Fatal("stream parity", err) + } + deadline := time.After(5 * time.Second) + for v.Status().DeltaSearch { + select { + case <-deadline: + t.Fatal("compaction did not run") + case <-time.After(20 * time.Millisecond): + } + } + if got, err := v.Find(ctx, text); err != nil || !reflect.DeepEqual(got, []string{"she", "한국어", "word-001"}) { + t.Fatal(got, err) + } +} diff --git a/pkg/acor/versioned_scale_test.go b/pkg/acor/versioned_scale_test.go index c349bcf..c8eb07a 100644 --- a/pkg/acor/versioned_scale_test.go +++ b/pkg/acor/versioned_scale_test.go @@ -40,7 +40,8 @@ func TestVersionedScale(t *testing.T) { } ctx := context.Background() name := "scale-" + v3ID() - opts := &VersionedOptions{Redis: AhoCorasickArgs{Addr: addr, Name: name}, PollInterval: 50 * time.Millisecond} + opts := &VersionedOptions{Redis: AhoCorasickArgs{Addr: addr, Name: name}, PollInterval: 50 * time.Millisecond, + DeltaSearch: os.Getenv("ACOR_V3_SCALE_DELTA") == "1"} v, err := OpenVersioned(ctx, opts) if err != nil { t.Fatal(err) @@ -95,10 +96,12 @@ func TestVersionedScale(t *testing.T) { start := time.Now() r, e := write() commit := time.Since(start) + committedAt := time.Now() if e == nil { e = v.WaitForVersion(ctx, r.Version) } ready := time.Since(start) + commitReady := time.Since(committedAt) close(done) wg.Wait() if e != nil { @@ -119,18 +122,20 @@ func TestVersionedScale(t *testing.T) { rss *= 1024 } record := map[string]interface{}{"operation": label, - "n": n, - "kind": kind, - "commit_ms": float64(commit.Microseconds()) / 1000, - "ready_ms": float64(ready.Microseconds()) / 1000, - "max_rss_bytes": rss, - "redis_bytes": after["used_memory"], - "sent_bytes": after["total_net_input_bytes"] - before["total_net_input_bytes"], - "received_bytes": after["total_net_output_bytes"] - before["total_net_output_bytes"], - "search_samples": len(latencies), - "search_p50_ns": percentile(50), - "search_p95_ns": percentile(95), - "search_p99_ns": percentile(99)} + "n": n, + "kind": kind, + "commit_ms": float64(commit.Microseconds()) / 1000, + "ready_ms": float64(ready.Microseconds()) / 1000, + "commit_to_ready_ms": float64(commitReady.Microseconds()) / 1000, + "delta_search": opts.DeltaSearch, + "max_rss_bytes": rss, + "redis_bytes": after["used_memory"], + "sent_bytes": after["total_net_input_bytes"] - before["total_net_input_bytes"], + "received_bytes": after["total_net_output_bytes"] - before["total_net_output_bytes"], + "search_samples": len(latencies), + "search_p50_ns": percentile(50), + "search_p95_ns": percentile(95), + "search_p99_ns": percentile(99)} data, _ := json.Marshal(record) t.Log(string(data)) return r @@ -183,6 +188,11 @@ func TestVersionedScale(t *testing.T) { t.Fatal("search differs from naive reference") } measure("identical_replace", func() (*WriteResult, error) { return v.Replace(ctx, r.Version, words) }) + for i := range 3 { + word := fmt.Sprintf("single-change-%d", i) + r = measure(fmt.Sprintf("add_1_repeat_%d", i), func() (*WriteResult, error) { return v.Add(ctx, r.Version, word) }) + r = measure(fmt.Sprintf("remove_1_repeat_%d", i), func() (*WriteResult, error) { return v.Remove(ctx, r.Version, word) }) + } for _, changes := range []int{1, 1000, max(1, n/100)} { added := make([]string, changes) for i := range added { From bfa17ef439ce4bfbe9dc0c4b42f2d551cca49439 Mon Sep 17 00:00:00 2001 From: Sung-Kyu Yoo Date: Sun, 13 Sep 2026 22:32:33 +0900 Subject: [PATCH 2/4] Fix delta overlay early termination and compaction locking --- internal/engine/overlay.go | 164 ++++++++++++++++++++------------ internal/engine/overlay_test.go | 28 ++++++ pkg/acor/versioned_overlay.go | 20 ++-- 3 files changed, 146 insertions(+), 66 deletions(-) diff --git a/internal/engine/overlay.go b/internal/engine/overlay.go index ebebc36..8d7a9ac 100644 --- a/internal/engine/overlay.go +++ b/internal/engine/overlay.go @@ -4,6 +4,7 @@ package engine import ( "slices" + "unicode/utf8" ) // Overlay combines an immutable base with a small replacement set. Its inputs @@ -29,81 +30,124 @@ type overlayMatch struct { func (o *overlayEngine) buildFromKeywords(map[string]struct{}) { panic("engine: immutable overlay") } func (o *overlayEngine) matchString(text string, emit func(string, int, int) bool) { - var matches []overlayMatch - order := 0 - o.base.MatchString(text, func(k string, start, end int) bool { - if _, deleted := o.removed[k]; !deleted { - matches = append(matches, overlayMatch{k, start, end, 0, order}) - } - order++ - return true - }) - order = 0 - o.added.MatchString(text, func(k string, start, end int) bool { - matches = append(matches, overlayMatch{k, start, end, 1, order}) - order++ - return true - }) - slices.SortStableFunc(matches, func(a, b overlayMatch) int { - if a.end != b.end { - return a.end - b.end + i := 0 + o.matchStream(func() (rune, bool) { + if i == len(text) { + return 0, false } - 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 - } - } + 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) { - const chunk = 4096 - longest := max(o.base.MaxKeywordRunes(), o.added.MaxKeywordRunes()) - window := make([]rune, 0, chunk+longest) - windowStart := 0 - total := 0 - stopped := false - emittedEnd := 0 - process := func() { - o.matchString(string(window), func(k string, from, end int) bool { - if windowStart+end <= emittedEnd { - return true - } - if !emit(k, windowStart+from, windowStart+end) { - stopped = true - return false + 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 }) - emittedEnd = total - keep := min(len(window), max(0, longest-1)) - windowStart = total - keep - copy(window, window[len(window)-keep:]) - window = window[:keep] + 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 { - r, ok := next() + ch, ok := next() if !ok { - break + return + } + end++ + window = append(window, ch) + if len(window) > longest { + copy(window, window[1:]) + window = window[:longest] } - window = append(window, r) - total++ - if len(window) >= chunk+longest-1 { - process() - if stopped { + 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 } } } - if len(window) > 0 && !stopped { - process() +} + +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 } } diff --git a/internal/engine/overlay_test.go b/internal/engine/overlay_test.go index bd39a32..57fecc5 100644 --- a/internal/engine/overlay_test.go +++ b/internal/engine/overlay_test.go @@ -45,3 +45,31 @@ func TestOverlayMatchesRebuild(t *testing.T) { 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) + } +} diff --git a/pkg/acor/versioned_overlay.go b/pkg/acor/versioned_overlay.go index 5057f0d..158ba6d 100644 --- a/pkg/acor/versioned_overlay.go +++ b/pkg/acor/versioned_overlay.go @@ -128,8 +128,6 @@ func (v *VersionedCollection) compactIfIdle(ctx context.Context) { if current == nil || current.base == nil || time.Since(current.installed) < v3CompactDelay { return } - v.refreshMu.Lock() - defer v.refreshMu.Unlock() if v.current.Load() != current || ctx.Err() != nil { return } @@ -140,26 +138,36 @@ func (v *VersionedCollection) compactIfIdle(ctx context.Context) { v.mu.Unlock() defer func() { v.mu.Lock() - v.status.Building = false - v.status.BuildDuration = time.Since(start) + if v.current.Load() == current { + v.status.Building = false + v.status.BuildDuration = time.Since(start) + } v.mu.Unlock() }() e := matchengine.New(enginePreset(v.opts.Preset)) if err := e.BuildSequenceContext(ctx, bucketSequence(current.buckets), current.manifest.Count); err != nil { - v.recordCompactError(err) + if v.current.Load() == current { + v.recordCompactError(err) + } return } active, err := v.client.Get(ctx, v.key("active")).Result() if err != nil { - v.recordCompactError(err) + if v.current.Load() == current { + v.recordCompactError(err) + } return } + v.refreshMu.Lock() + defer v.refreshMu.Unlock() if Version(active) != current.version || v.current.Load() != current || ctx.Err() != nil { return } v.current.Store(&v3Engine{engine: e, version: current.version, sequence: current.sequence, manifest: current.manifest, buckets: current.buckets, installed: time.Now()}) v.mu.Lock() + v.status.Building = false + v.status.BuildDuration = time.Since(start) v.status.LastError = "" v.status.CompletedBuilds++ v.status.DeltaSearch = false From 4bcbf6054fb60924978af07eca98779f8b934a35 Mon Sep 17 00:00:00 2001 From: Sung-Kyu Yoo Date: Sun, 13 Sep 2026 22:39:24 +0900 Subject: [PATCH 3/4] Cover overlay compaction Redis failure --- pkg/acor/versioned_overlay_test.go | 39 ++++++++++++++++++++++++++++++ 1 file changed, 39 insertions(+) diff --git a/pkg/acor/versioned_overlay_test.go b/pkg/acor/versioned_overlay_test.go index 8d7ee5d..a169216 100644 --- a/pkg/acor/versioned_overlay_test.go +++ b/pkg/acor/versioned_overlay_test.go @@ -78,3 +78,42 @@ func TestVersionedOverlayLifecycle(t *testing.T) { t.Fatal(got, err) } } + +func TestVersionedOverlayCompactionRedisError(t *testing.T) { + ctx := context.Background() + server := miniredis.RunT(t) + v, err := OpenVersioned(ctx, &VersionedOptions{ + Redis: AhoCorasickArgs{Addr: server.Addr(), Name: "overlay-compact-error"}, + DeltaSearch: true, PollInterval: time.Hour, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = v.Close() }) + base := make([]string, 129) + for i := range base { + base[i] = fmt.Sprintf("word-%03d", i) + } + r, err := v.Replace(ctx, v.Status().ServingVersion, base) + if err != nil { + t.Fatal(err) + } + waitV3(t, v, r.Version) + r, err = v.Add(ctx, r.Version, "new-word") + if err != nil { + t.Fatal(err) + } + waitV3(t, v, r.Version) + current := v.current.Load() + if current.base == nil { + t.Fatal("expected an overlay") + } + ready := *current + ready.installed = time.Now().Add(-v3CompactDelay) + v.current.Store(&ready) + server.Close() + v.compactIfIdle(ctx) + if s := v.Status(); s.LastError == "" || s.ServingVersion != r.Version || !s.DeltaSearch { + t.Fatalf("failed compaction changed serving snapshot: %+v", s) + } +} From 097b7f2a0b5447585e89dbefaa9b7af774b09da0 Mon Sep 17 00:00:00 2001 From: Sung-Kyu Yoo Date: Sun, 13 Sep 2026 22:49:06 +0900 Subject: [PATCH 4/4] Link delta search changelog entry to PR 255 --- changes/unreleased/20260913-v3-delta-search.yaml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/changes/unreleased/20260913-v3-delta-search.yaml b/changes/unreleased/20260913-v3-delta-search.yaml index 7b7d76c..c3d4e6f 100644 --- a/changes/unreleased/20260913-v3-delta-search.yaml +++ b/changes/unreleased/20260913-v3-delta-search.yaml @@ -1,3 +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"