amortize capped Source.Append to O(1)

source.go

@@ -26,11 +26,16 @@ type Source struct {
 	inClosed   bool
 	isInfinite bool
 	lines      []line.Line
-	name       string
-	mutex      sync.RWMutex
-	ready      chan struct{}
-	setupDone  chan struct{}
-	setupOnce  sync.Once
+	// start is the index of the oldest live line within lines. When a
+	// capacity is set, Append advances start instead of reallocating on
+	// every line; the dead prefix lines[:start] is reclaimed in bulk by a
+	// periodic compaction. The live window is always lines[start:].
+	start     int
+	name      string
+	mutex     sync.RWMutex
+	ready     chan struct{}
+	setupDone chan struct{}
+	setupOnce sync.Once
 }
 
 // drawRefreshInterval is the interval at which the screen is redrawn while
@@ -225,7 +230,12 @@ func (s *Source) Start(ctx context.Context, out pipeline.ChanOutput) {
 
 	if !resume {
 		// no fancy resume handling needed. Send individual lines.
-		for _, l := range s.lines {
+		// setupDone is closed, so the buffer is stable; snapshot the live
+		// window (lines[start:]) under the lock and iterate it.
+		s.mutex.RLock()
+		live := s.lines[s.start:]
+		s.mutex.RUnlock()
+		for _, l := range live {
 			select {
 			case <-ctx.Done():
 				if pdebug.Enabled {
@@ -314,40 +324,60 @@ func (s *Source) SetupDone() <-chan struct{} {
 	return s.setupDone
 }
 
-// linesInRange returns a slice of lines between start and end indices from the buffer.
-func (s *Source) linesInRange(start, end int) []line.Line {
+// linesInRange returns a slice of lines between from and to (indices into the
+// live window) from the buffer.
+func (s *Source) linesInRange(from, to int) []line.Line {
 	s.mutex.RLock()
 	defer s.mutex.RUnlock()
-	return s.lines[start:end]
+	return s.lines[s.start+from : s.start+to]
 }
 
-// LineAt returns the line at the given index from the buffer.
+// LineAt returns the line at the given index (within the live window) from the buffer.
 func (s *Source) LineAt(n int) (line.Line, error) {
 	s.mutex.RLock()
 	defer s.mutex.RUnlock()
-	return bufferLineAt(s.lines, n)
+	return bufferLineAt(s.lines[s.start:], n)
 }
 
 // Size returns the number of lines currently in the buffer.
 func (s *Source) Size() int {
 	s.mutex.RLock()
 	defer s.mutex.RUnlock()
-	return len(s.lines)
+	return len(s.lines) - s.start
 }
 
 // Append adds a new line to the source buffer. If a capacity is set and
 // exceeded, the oldest lines are discarded to maintain the limit.
+//
+// Discarding is amortized O(1): rather than reallocating and copying the
+// whole window on every line once saturated (which is O(capacity) per
+// Append), we advance the logical start index and only compact — copying
+// the live window to the front and releasing the discarded lines — once the
+// dead prefix has grown to a full window. Compaction therefore happens once
+// every capacity appends, so the per-line cost is O(1) amortized while peak
+// memory stays bounded at ~2*capacity lines.
 func (s *Source) Append(l line.Line) {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
 	s.lines = append(s.lines, l)
-	if s.capacity > 0 && len(s.lines) > s.capacity {
-		diff := len(s.lines) - s.capacity
+	if s.capacity <= 0 {
+		return
+	}
 
-		// Copy to a new slice to allow GC of discarded lines
-		newLines := make([]line.Line, s.capacity)
-		copy(newLines, s.lines[diff:])
-		s.lines = newLines
+	// Drop the oldest line logically once we are over capacity. len-start
+	// is the live size, so this keeps Size() pinned at exactly capacity.
+	if len(s.lines)-s.start > s.capacity {
+		s.start++
+	}
+
+	// Once the dead prefix is as large as the live window (len == 2*capacity),
+	// compact: slide the live window to the front, clear the freed tail so the
+	// discarded lines can be GC'd, and reset start.
+	if s.start >= s.capacity {
+		n := copy(s.lines, s.lines[s.start:])
+		clear(s.lines[n:])
+		s.lines = s.lines[:n]
+		s.start = 0
 	}
 }

source_capacity_bench_test.go

@@ -0,0 +1,30 @@
+package peco
+
+import (
+	"strconv"
+	"strings"
+	"testing"
+
+	"github.com/peco/peco/line"
+)
+
+// BenchmarkSourceAppendSaturated measures Append cost on a capacity-bounded
+// source that is already saturated, which is the steady state for a long-lived
+// streaming input (e.g. `tail -f | peco -b N`).
+func BenchmarkSourceAppendSaturated(b *testing.B) {
+	const capacity = 10000
+	ig := newIDGen()
+	go ig.Run(b.Context())
+
+	s := NewSource("-", strings.NewReader(""), false, ig, capacity, false, false)
+	// Pre-fill to capacity so every benchmarked Append discards an old line.
+	for i := range capacity {
+		s.Append(line.NewRaw(uint64(i), strconv.Itoa(i), false, false))
+	}
+
+	l := line.NewRaw(uint64(capacity), "payload", false, false)
+	b.ResetTimer()
+	for range b.N {
+		s.Append(l)
+	}
+}

source_capacity_test.go

@@ -0,0 +1,76 @@
+package peco
+
+import (
+	"fmt"
+	"strings"
+	"testing"
+
+	"github.com/peco/peco/line"
+	"github.com/stretchr/testify/require"
+)
+
+// TestSourceCapacity verifies that a capacity-bounded Source keeps exactly the
+// most recent `capacity` lines as new lines stream in, across many compaction
+// boundaries, and that the backing storage stays bounded (amortized O(1)
+// trimming rather than a full reallocation per append).
+func TestSourceCapacity(t *testing.T) {
+	const capacity = 4
+	ig := newIDGen()
+	go ig.Run(t.Context())
+
+	s := NewSource("-", strings.NewReader(""), false, ig, capacity, false, false)
+
+	const total = 25 // crosses several capacity-sized compaction windows
+	for i := range total {
+		s.Append(line.NewRaw(uint64(i), fmt.Sprintf("line%d", i), false, false))
+
+		want := min(i+1, capacity)
+		require.Equal(t, want, s.Size(), "Size must stay pinned at capacity once saturated")
+
+		// The live window must be exactly the most-recently-appended lines.
+		oldest := (i + 1) - s.Size()
+		for j := range s.Size() {
+			l, err := s.LineAt(j)
+			require.NoError(t, err, "LineAt(%d) at append %d", j, i)
+			require.Equal(t, fmt.Sprintf("line%d", oldest+j), l.DisplayString())
+		}
+
+		// Backing storage must not grow without bound.
+		require.LessOrEqual(t, len(s.lines), 2*capacity,
+			"backing window should stay bounded at ~2*capacity")
+	}
+
+	// linesInRange over the live window still returns a correct contiguous slice
+	// after many compactions.
+	rng := s.linesInRange(0, capacity)
+	require.Len(t, rng, capacity)
+	require.Equal(t, fmt.Sprintf("line%d", total-capacity), rng[0].DisplayString())
+	require.Equal(t, fmt.Sprintf("line%d", total-1), rng[capacity-1].DisplayString())
+
+	// Out-of-range access is rejected.
+	_, err := s.LineAt(capacity)
+	require.Error(t, err)
+}
+
+// TestSourceUnlimitedRetainsAll verifies that the default (capacity 0) keeps
+// every appended line — the start-offset machinery must not kick in.
+func TestSourceUnlimitedRetainsAll(t *testing.T) {
+	ig := newIDGen()
+	go ig.Run(t.Context())
+
+	s := NewSource("-", strings.NewReader(""), false, ig, 0, false, false)
+
+	const total = 50
+	for i := range total {
+		s.Append(line.NewRaw(uint64(i), fmt.Sprintf("line%d", i), false, false))
+	}
+
+	require.Equal(t, total, s.Size())
+	require.Equal(t, 0, s.start, "start must stay 0 when capacity is unlimited")
+	first, err := s.LineAt(0)
+	require.NoError(t, err)
+	require.Equal(t, "line0", first.DisplayString())
+	last, err := s.LineAt(total - 1)
+	require.NoError(t, err)
+	require.Equal(t, fmt.Sprintf("line%d", total-1), last.DisplayString())
+}