From 311e71399cc6f32d1c4cd23fc37db17d18e0af0c Mon Sep 17 00:00:00 2001 From: Lakshman Patel Date: Tue, 14 Jul 2026 12:27:49 +0530 Subject: [PATCH 1/3] fix(stream): prevent Reset()/backgroundCompress race and double-Close panic StreamCompressor.Reset() could be clobbered by a stale in-flight backgroundCompress goroutine writing its result after the reset, and Close() panicked if called twice due to an unguarded close(done). Add an epoch counter that Reset() bumps; backgroundCompress captures the epoch when it starts and discards its result if the epoch has since changed, making stale writes a no-op instead of an overwrite. Guard Close() with sync.Once so repeated calls are safe. Co-Authored-By: Claude Sonnet 5 --- stream.go | 34 ++++++++++++++++++++++++++------- stream_test.go | 52 ++++++++++++++++++++++++++++++++++++++++++++++++++ 2 files changed, 79 insertions(+), 7 deletions(-) diff --git a/stream.go b/stream.go index 5cfb35e3e..4bfb41487 100644 --- a/stream.go +++ b/stream.go @@ -25,7 +25,9 @@ type StreamCompressor struct { dirty bool // new content since last compression compressing bool // background compression in progress lastCompressedIdx int // raw[] index up to which content has been compressed + epoch uint64 // bumped on Reset; stale background results are discarded done chan struct{} + closeOnce sync.Once wg sync.WaitGroup // tracks in-progress compression goroutines } @@ -63,6 +65,7 @@ func (sc *StreamCompressor) Append(content string) { // Take a snapshot of the delta (segments not yet compressed) under lock. var deltaCopy string var rawLen int + epoch := sc.epoch if shouldCompress { rawLen = len(sc.raw) deltaIdx := sc.lastCompressedIdx @@ -78,20 +81,25 @@ func (sc *StreamCompressor) Append(content string) { if shouldCompress { sc.wg.Add(1) - go sc.backgroundCompress(deltaCopy, rawLen) + go sc.backgroundCompress(deltaCopy, rawLen, epoch) } } // backgroundCompress runs delta-only compression on the given content and -// merges the result with the existing compressed output. -func (sc *StreamCompressor) backgroundCompress(delta string, compressedUpTo int) { +// merges the result with the existing compressed output. epoch is the +// generation captured when the goroutine was started; if Reset() has since +// bumped the generation, the result is stale and is discarded instead of +// being written back over the freshly-reset state. +func (sc *StreamCompressor) backgroundCompress(delta string, compressedUpTo int, epoch uint64) { defer sc.wg.Done() // Check if we've been shut down before starting. select { case <-sc.done: sc.mu.Lock() - sc.compressing = false + if sc.epoch == epoch { + sc.compressing = false + } sc.mu.Unlock() return default: @@ -103,6 +111,12 @@ func (sc *StreamCompressor) backgroundCompress(delta string, compressedUpTo int) sc.mu.Lock() defer sc.mu.Unlock() + // If Reset() ran while we were compressing, this result belongs to a + // stale generation; discard it rather than overwriting the reset state. + if sc.epoch != epoch { + return + } + // Merge: accumulate compressed output rather than replacing it. if sc.compressed == "" { sc.compressed = compressed @@ -150,11 +164,15 @@ func (sc *StreamCompressor) TokenCount() int { } // Reset clears all accumulated content and resets the compressor to its -// initial state, allowing it to be reused. +// initial state, allowing it to be reused. It bumps the internal generation +// counter so that any backgroundCompress goroutine still in flight from +// before the reset discards its result instead of overwriting the freshly +// reset state. func (sc *StreamCompressor) Reset() { sc.mu.Lock() defer sc.mu.Unlock() + sc.epoch++ sc.raw = nil sc.compressed = "" sc.stats = Stats{} @@ -164,9 +182,11 @@ func (sc *StreamCompressor) Reset() { } // Close shuts down the background compressor and waits for any in-progress -// compression to finish. +// compression to finish. Safe to call multiple times. func (sc *StreamCompressor) Close() { - close(sc.done) + sc.closeOnce.Do(func() { + close(sc.done) + }) sc.wg.Wait() } diff --git a/stream_test.go b/stream_test.go index 215cd35c9..a126f6dba 100644 --- a/stream_test.go +++ b/stream_test.go @@ -298,6 +298,58 @@ func TestStreamCompressor_Reset(t *testing.T) { sc.Close() } +func TestStreamCompressor_ResetRacesBackgroundCompress(t *testing.T) { + sc := tok.NewStreamCompressor(5, tok.Aggressive) + defer sc.Close() + + for i := 0; i < 50; i++ { + // Trigger a background compression, then immediately race a Reset + // against it. The background goroutine's result must never land + // after the reset (i.e. post-reset state must stay empty/consistent, + // not get clobbered by a stale write once the goroutine finishes). + sc.Append(strings.Repeat("hello world reset race test content. ", 20)) + + var wg sync.WaitGroup + wg.Add(1) + go func() { + defer wg.Done() + sc.Reset() + }() + wg.Wait() + + if raw := sc.Raw(); raw != "" { + t.Fatalf("iteration %d: expected empty raw immediately after Reset(), got %q", i, raw) + } + + // Give any stale background goroutine time to finish and (if the + // bug were present) write its result back after the reset. + time.Sleep(2 * time.Millisecond) + + if raw := sc.Raw(); raw != "" { + t.Fatalf("iteration %d: stale background compress overwrote reset state, raw=%q", i, raw) + } + if snap, stats := sc.Snapshot(); snap != "" || stats.OriginalTokens != 0 { + t.Fatalf("iteration %d: stale background compress overwrote reset state, snapshot=%q stats=%+v", i, snap, stats) + } + } +} + +func TestStreamCompressor_CloseTwiceDoesNotPanic(t *testing.T) { + sc := tok.NewStreamCompressor(5, tok.Minimal) + + sc.Append(strings.Repeat("close twice test content. ", 20)) + + sc.Close() + + // Second Close must be a safe no-op, not a panic on closing a closed channel. + defer func() { + if r := recover(); r != nil { + t.Fatalf("second Close() panicked: %v", r) + } + }() + sc.Close() +} + func TestStreamCompressor_DeltaStatsAccumulate(t *testing.T) { sc := tok.NewStreamCompressor(5, tok.Minimal) defer sc.Close() From ef5a2aa8b9caacf3c66223e895d80d97ca50df40 Mon Sep 17 00:00:00 2001 From: Lakshman Patel Date: Tue, 14 Jul 2026 14:44:23 +0530 Subject: [PATCH 2/3] chore: remove parallel filter file --- internal/filter/parallel.go | 339 ------------------------------------ 1 file changed, 339 deletions(-) delete mode 100644 internal/filter/parallel.go diff --git a/internal/filter/parallel.go b/internal/filter/parallel.go deleted file mode 100644 index 88da13793..000000000 --- a/internal/filter/parallel.go +++ /dev/null @@ -1,339 +0,0 @@ -package filter - -import ( - "context" - "runtime" - "sync" -) - -// ParallelFilterResult holds result from parallel filter execution -type ParallelFilterResult struct { - Output string - Saved int - Error error -} - -// ExecuteFiltersParallel runs independent filters in parallel -// This improves throughput for multi-core systems -func ExecuteFiltersParallel(filters []filterLayer, input string, mode Mode) (string, int) { - if len(filters) == 0 { - return input, 0 - } - - // Single filter: run directly (avoid goroutine overhead) - if len(filters) == 1 { - return filters[0].filter.Apply(input, mode) - } - - // Parallel execution for multiple filters - results := make([]ParallelFilterResult, len(filters)) - var wg sync.WaitGroup - - // Run filters in parallel - for i, layer := range filters { - wg.Add(1) - go func(idx int, l filterLayer) { - defer wg.Done() - output, saved := l.filter.Apply(input, mode) - results[idx] = ParallelFilterResult{ - Output: output, - Saved: saved, - } - }(i, layer) - } - - wg.Wait() - - // Return best result (most savings) and its corresponding savings. - // Filters run independently on the same input; we pick the best output. - bestResult := results[0] - for _, r := range results { - if r.Saved > bestResult.Saved { - bestResult = r - } - } - - return bestResult.Output, bestResult.Saved -} - -// ExecuteFiltersSequential runs filters sequentially -// Use when filters depend on each other's output -func ExecuteFiltersSequential(filters []filterLayer, input string, mode Mode) (string, int) { - output := input - totalSaved := 0 - - for _, layer := range filters { - newOutput, saved := layer.filter.Apply(output, mode) - output = newOutput - totalSaved += saved - } - - return output, totalSaved -} - -// ShouldUseParallel determines if parallel execution is beneficial -func ShouldUseParallel(filters []filterLayer, inputSize int) bool { - // Use parallel for: - // - Multiple filters (2+) - // - Large inputs (>1KB) - // - Independent filters - - if len(filters) < 2 { - return false - } - - if inputSize < 1024 { - return false // Overhead not worth it for small inputs - } - - // Check if we have enough CPU cores - // runtime.NumCPU() >= 2 - - return true -} - -// ParallelPipelineStats holds stats from parallel execution -type ParallelPipelineStats struct { - mu sync.Mutex - LayerStats map[string]LayerStat - TotalSaved int - ParallelTime int64 - SequentialTime int64 -} - -// NewParallelPipelineStats creates new stats tracker -func NewParallelPipelineStats() *ParallelPipelineStats { - return &ParallelPipelineStats{ - LayerStats: make(map[string]LayerStat), - } -} - -// AddStat adds a layer stat thread-safely -func (s *ParallelPipelineStats) AddStat(name string, stat LayerStat) { - s.mu.Lock() - defer s.mu.Unlock() - s.LayerStats[name] = stat - s.TotalSaved += stat.TokensSaved -} - -// ParallelProcessor handles parallel compression of multiple inputs -// Uses worker pool pattern for optimal CPU utilization -type ParallelProcessor struct { - workers int - sem chan struct{} -} - -// NewParallelProcessor creates a new parallel processor -// Automatically determines optimal worker count based on CPU cores -func NewParallelProcessor() *ParallelProcessor { - workers := runtime.NumCPU() - if workers < 4 { - workers = 4 - } - return &ParallelProcessor{ - workers: workers, - sem: make(chan struct{}, workers), - } -} - -// NewParallelProcessorWithWorkers creates processor with specific worker count -func NewParallelProcessorWithWorkers(workers int) *ParallelProcessor { - if workers < 1 { - workers = 1 - } - if workers > runtime.NumCPU()*4 { - workers = runtime.NumCPU() * 4 - } - return &ParallelProcessor{ - workers: workers, - sem: make(chan struct{}, workers), - } -} - -// ProcessItems processes multiple items in parallel -// Each item is processed by the provided function -func (p *ParallelProcessor) ProcessItems(items []string, processFn func(string) (string, int)) []ParallelProcessResult { - if len(items) == 0 { - return nil - } - - // For small batches, process sequentially (avoid overhead) - if len(items) < 4 { - results := make([]ParallelProcessResult, len(items)) - for i, item := range items { - output, saved := processFn(item) - results[i] = ParallelProcessResult{ - Input: item, - Output: output, - Saved: saved, - Index: i, - } - } - return results - } - - // Process in parallel - results := make([]ParallelProcessResult, len(items)) - var wg sync.WaitGroup - - for i, item := range items { - wg.Add(1) - p.sem <- struct{}{} // Acquire semaphore - - go func(index int, input string) { - defer wg.Done() - defer func() { <-p.sem }() // Release semaphore - - output, saved := processFn(input) - results[index] = ParallelProcessResult{ - Input: input, - Output: output, - Saved: saved, - Index: index, - } - }(i, item) - } - - wg.Wait() - return results -} - -// ProcessItemsContext processes items with context cancellation support -func (p *ParallelProcessor) ProcessItemsContext(ctx context.Context, items []string, processFn func(context.Context, string) (string, int)) ([]ParallelProcessResult, error) { - if len(items) == 0 { - return nil, nil - } - - // Check context before starting - select { - case <-ctx.Done(): - return nil, ctx.Err() - default: - } - - results := make([]ParallelProcessResult, len(items)) - var wg sync.WaitGroup - errCh := make(chan error, 1) - - for i, item := range items { - wg.Add(1) - - go func(index int, input string) { - defer wg.Done() - - // Acquire semaphore - select { - case p.sem <- struct{}{}: - case <-ctx.Done(): - select { - case errCh <- ctx.Err(): - default: - } - return - } - defer func() { <-p.sem }() - - select { - case <-ctx.Done(): - select { - case errCh <- ctx.Err(): - default: - } - return - default: - } - - output, saved := processFn(ctx, input) - results[index] = ParallelProcessResult{ - Input: input, - Output: output, - Saved: saved, - Index: index, - } - }(i, item) - } - - wg.Wait() - close(errCh) - - if err := <-errCh; err != nil { - return results, err - } - return results, nil -} - -// ParallelProcessResult holds the result of a parallel processing operation -type ParallelProcessResult struct { - Input string - Output string - Saved int - Index int -} - -// TotalSaved calculates total tokens saved from results -func TotalSaved(results []ParallelProcessResult) int { - total := 0 - for _, r := range results { - total += r.Saved - } - return total -} - -// CollectOutputs collects all outputs from results in order -func CollectOutputs(results []ParallelProcessResult) []string { - outputs := make([]string, len(results)) - for _, r := range results { - outputs[r.Index] = r.Output - } - return outputs -} - -// ParallelCompressor provides high-level parallel compression interface -type ParallelCompressor struct { - processor *ParallelProcessor - engine *PipelineCoordinator -} - -// NewParallelCompressor creates a new parallel compressor -func NewParallelCompressor(config PipelineConfig) *ParallelCompressor { - return &ParallelCompressor{ - processor: NewParallelProcessor(), - engine: NewPipelineCoordinator(&config), - } -} - -// Compress compresses a single input -func (pc *ParallelCompressor) Compress(input string) (string, int) { - output, stats := pc.engine.Process(input) - if stats == nil { - return output, 0 - } - return output, stats.TotalSaved -} - -// CompressBatch compresses multiple inputs in parallel -func (pc *ParallelCompressor) CompressBatch(inputs []string) []ParallelProcessResult { - return pc.processor.ProcessItems(inputs, func(item string) (string, int) { - output, stats := pc.engine.Process(item) - if stats == nil { - return output, 0 - } - return output, stats.TotalSaved - }) -} - -// CompressBatchContext compresses with context support -func (pc *ParallelCompressor) CompressBatchContext(ctx context.Context, inputs []string) ([]ParallelProcessResult, error) { - return pc.processor.ProcessItemsContext(ctx, inputs, func(ctx context.Context, item string) (string, int) { - output, stats := pc.engine.Process(item) - if stats == nil { - return output, 0 - } - return output, stats.TotalSaved - }) -} - -// WorkerCount returns the number of workers -func (pc *ParallelCompressor) WorkerCount() int { - return pc.processor.workers -} From 67e5849e5b9068bf8f020854fe4cfc40693f9382 Mon Sep 17 00:00:00 2001 From: Lakshman Patel Date: Tue, 14 Jul 2026 15:01:39 +0530 Subject: [PATCH 3/3] fix: remove benchmark referencing deleted parallel processor --- .../filter/optimizations_benchmark_test.go | 35 ------------------- 1 file changed, 35 deletions(-) diff --git a/internal/filter/optimizations_benchmark_test.go b/internal/filter/optimizations_benchmark_test.go index 63514b896..5929a57d2 100644 --- a/internal/filter/optimizations_benchmark_test.go +++ b/internal/filter/optimizations_benchmark_test.go @@ -155,41 +155,6 @@ func BenchmarkFastStringBuilder(b *testing.B) { } } -// BenchmarkParallelProcessor benchmarks parallel processing -func BenchmarkParallelProcessor(b *testing.B) { - processor := NewParallelProcessor() - - // Simple processing function - processFn := func(input string) (string, int) { - // Simple transformation: uppercase - output := fastops.FastLower(input) - return output, len(input) - len(output) - } - - sizes := []int{10, 100, 1000} - - for _, size := range sizes { - items := make([]string, size) - for i := range items { - items[i] = makeString(100) - } - - b.Run(fmt.Sprintf("Parallel_%d_items", size), func(b *testing.B) { - for i := 0; i < b.N; i++ { - processor.ProcessItems(items, processFn) - } - }) - - b.Run(fmt.Sprintf("Sequential_%d_items", size), func(b *testing.B) { - for i := 0; i < b.N; i++ { - for _, item := range items { - processFn(item) - } - } - }) - } -} - // BenchmarkPipeline benchmarks the full pipeline func BenchmarkPipeline(b *testing.B) { configs := []struct {