progress: fix data races on ticker, states, spinner, and bar state (#17445)

* progress: fix data races on ticker, states, spinner, and bar state

NewProgress spawned start() which wrote p.ticker while stop() read and
cleared it with no synchronization; stop() and StopAndClear() also read
p.states and p.pos outside p.mu, Spinner's start() goroutine raced
Stop() and String() on s.value/s.stopped/s.ticker, and Bar.Set raced
Bar.String on currentValue/stopped/buckets (callback goroutine vs the
render goroutine). Detected by go test -race across cmd and cmd/launch
(~20 warnings; the Bar race is latent — never flagged because tests
don't interleave it, but real in production pull/push progress).

Create tickers before spawning the render goroutines and pass the
channel in, guard Progress internals with p.mu throughout stop() (via a
renderLocked core), and give Spinner and Bar their own mutexes.

* use a more idiomatic channel based done signal

progress/bar.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"os"
 	"strings"
+	"sync"
 	"time"
 
 	"golang.org/x/term"
@@ -12,6 +13,9 @@ import (
 )
 
 type Bar struct {
+	// mu guards all fields below: Set is called from download progress
+	// callbacks while String is called from the Progress render goroutine.
+	mu           sync.Mutex
 	message      string
 	messageWidth int
 
@@ -67,6 +71,9 @@ func (b *Bar) String() string {
 		termWidth = defaultTermWidth
 	}
 
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
 	var pre strings.Builder
 	if len(b.message) > 0 {
 		message := strings.TrimSpace(b.message)
@@ -150,6 +157,9 @@ func (b *Bar) String() string {
 }
 
 func (b *Bar) Set(value int64) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+
 	if value >= b.maxValue {
 		value = b.maxValue
 	}
@@ -172,6 +182,7 @@ func (b *Bar) Set(value int64) {
 	}
 }
 
+// percent must be called with b.mu held.
 func (b *Bar) percent() float64 {
 	if b.maxValue > 0 {
 		return float64(b.currentValue) / float64(b.maxValue) * 100
@@ -180,6 +191,7 @@ func (b *Bar) percent() float64 {
 	return 0
 }
 
+// rate must be called with b.mu held.
 func (b *Bar) rate() float64 {
 	var numerator, denominator float64
 

progress/progress.go

@@ -27,35 +27,45 @@ type Progress struct {
 
 	pos int
 
-	ticker *time.Ticker
 	states []State
+
+	stopOnce sync.Once
+	// done is closed to tell the render loop to exit.
+	done chan struct{}
 }
 
 func NewProgress(w io.Writer) *Progress {
-	p := &Progress{w: bufio.NewWriter(w)}
+	p := &Progress{w: bufio.NewWriter(w), done: make(chan struct{})}
 	go p.start()
 	return p
 }
 
-func (p *Progress) stop() bool {
+// stop halts the render loop, stopping any spinners first. It reports whether
+// rendering was active and how many lines were last rendered.
+func (p *Progress) stop() (bool, int) {
+	var stopped bool
+	p.stopOnce.Do(func() {
+		close(p.done)
+		stopped = true
+	})
+
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
 	for _, state := range p.states {
 		if spinner, ok := state.(*Spinner); ok {
 			spinner.Stop()
 		}
 	}
 
-	if p.ticker != nil {
-		p.ticker.Stop()
-		p.ticker = nil
-		p.render()
-		return true
+	if stopped {
+		p.renderLocked()
 	}
-
-	return false
+	return stopped, p.pos
 }
 
 func (p *Progress) Stop() bool {
-	stopped := p.stop()
+	stopped, _ := p.stop()
 	if stopped {
 		fmt.Fprint(p.w, "\n")
 		p.w.Flush()
@@ -69,10 +79,10 @@ func (p *Progress) StopAndClear() bool {
 	fmt.Fprint(p.w, "\033[?25l")
 	defer fmt.Fprint(p.w, "\033[?25h")
 
-	stopped := p.stop()
+	stopped, pos := p.stop()
 	if stopped {
 		// clear all progress lines
-		for i := range p.pos {
+		for i := range pos {
 			if i > 0 {
 				fmt.Fprint(p.w, "\033[A")
 			}
@@ -91,14 +101,19 @@ func (p *Progress) Add(key string, state State) {
 }
 
 func (p *Progress) render() {
+	p.mu.Lock()
+	defer p.mu.Unlock()
+
+	p.renderLocked()
+}
+
+// renderLocked renders with p.mu held.
+func (p *Progress) renderLocked() {
 	_, termHeight, err := term.GetSize(int(os.Stderr.Fd()))
 	if err != nil {
 		termHeight = defaultTermHeight
 	}
 
-	p.mu.Lock()
-	defer p.mu.Unlock()
-
 	defer p.w.Flush()
 
 	// eliminate flickering on terminals that support synchronized output
@@ -127,8 +142,15 @@ func (p *Progress) render() {
 }
 
 func (p *Progress) start() {
-	p.ticker = time.NewTicker(100 * time.Millisecond)
-	for range p.ticker.C {
-		p.render()
+	ticker := time.NewTicker(100 * time.Millisecond)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-p.done:
+			return
+		case <-ticker.C:
+			p.render()
+		}
 	}
 }

progress/progress_test.go

@@ -0,0 +1,92 @@
+package progress
+
+import (
+	"bytes"
+	"runtime"
+	"testing"
+	"time"
+)
+
+// TestStopIsIdempotent pins the contract cmd relies on: a second Stop (e.g. a
+// deferred StopAndClear after an explicit one) reports false so callers don't
+// emit trailing output twice.
+func TestStopIsIdempotent(t *testing.T) {
+	for _, tt := range []struct {
+		name string
+		stop func(*Progress) bool
+	}{
+		{"Stop", (*Progress).Stop},
+		{"StopAndClear", (*Progress).StopAndClear},
+	} {
+		t.Run(tt.name, func(t *testing.T) {
+			p := NewProgress(&bytes.Buffer{})
+			if !tt.stop(p) {
+				t.Fatal("first stop should report true")
+			}
+			if tt.stop(p) {
+				t.Fatal("second stop should report false")
+			}
+		})
+	}
+}
+
+// TestStopStopsRendering verifies the render loop exits, so no further output
+// is written after Stop returns.
+func TestStopStopsRendering(t *testing.T) {
+	var buf bytes.Buffer
+	p := NewProgress(&buf)
+	p.Add("bar", NewBar("test", 100, 0))
+	time.Sleep(250 * time.Millisecond) // let a few ticks render
+	p.Stop()
+
+	settled := buf.Len()
+	time.Sleep(300 * time.Millisecond) // several tick intervals
+	if buf.Len() != settled {
+		t.Fatalf("render loop still writing after Stop: %d -> %d bytes", settled, buf.Len())
+	}
+}
+
+// TestStopReapsGoroutines verifies Stop reaps the render and spinner
+// goroutines rather than leaving them parked on a ticker. Spinners previously
+// ran until the process exited, so a long-lived `ollama run` leaked one per
+// progress bar it displayed.
+func TestStopReapsGoroutines(t *testing.T) {
+	base := runtime.NumGoroutine()
+
+	const cycles = 50
+	for range cycles {
+		p := NewProgress(&bytes.Buffer{})
+		p.Add("spin", NewSpinner("working"))
+		p.Add("bar", NewBar("dl", 100, 0))
+		p.Stop()
+	}
+
+	// Allow a moment for the goroutines to be descheduled before counting.
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) && runtime.NumGoroutine() > base+2 {
+		time.Sleep(50 * time.Millisecond)
+	}
+
+	if got := runtime.NumGoroutine(); got > base+2 {
+		t.Fatalf("goroutines after %d start/stop cycles: got %d, want <= %d", cycles, got, base+2)
+	}
+}
+
+// TestSpinnerStopStopsAnimating verifies Spinner.Stop halts its goroutine.
+func TestSpinnerStopStopsAnimating(t *testing.T) {
+	s := NewSpinner("working")
+	time.Sleep(250 * time.Millisecond)
+	s.Stop()
+
+	s.mu.Lock()
+	settled := s.value
+	s.mu.Unlock()
+
+	time.Sleep(300 * time.Millisecond)
+
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	if s.value != settled {
+		t.Fatalf("spinner still animating after Stop: %d -> %d", settled, s.value)
+	}
+}

progress/spinner.go

@@ -3,21 +3,27 @@ package progress
 import (
 	"fmt"
 	"strings"
+	"sync"
 	"sync/atomic"
 	"time"
 )
 
 type Spinner struct {
-	message      atomic.Value
+	message atomic.Value
+
+	mu           sync.Mutex
 	messageWidth int
 
 	parts []string
 
 	value int
 
-	ticker  *time.Ticker
 	started time.Time
 	stopped time.Time
+
+	stopOnce sync.Once
+	// done is closed to tell the animation loop to exit.
+	done chan struct{}
 }
 
 func NewSpinner(message string) *Spinner {
@@ -26,6 +32,7 @@ func NewSpinner(message string) *Spinner {
 			"⠋", "⠙", "⠹", "⠸", "⠼", "⠴", "⠦", "⠧", "⠇", "⠏",
 		},
 		started: time.Now(),
+		done:    make(chan struct{}),
 	}
 	s.SetMessage(message)
 	go s.start()
@@ -37,6 +44,9 @@ func (s *Spinner) SetMessage(message string) {
 }
 
 func (s *Spinner) String() string {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+
 	var sb strings.Builder
 
 	if message, ok := s.message.Load().(string); ok && len(message) > 0 {
@@ -63,17 +73,27 @@ func (s *Spinner) String() string {
 }
 
 func (s *Spinner) start() {
-	s.ticker = time.NewTicker(100 * time.Millisecond)
-	for range s.ticker.C {
-		s.value = (s.value + 1) % len(s.parts)
-		if !s.stopped.IsZero() {
+	ticker := time.NewTicker(100 * time.Millisecond)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-s.done:
 			return
+		case <-ticker.C:
+			s.mu.Lock()
+			s.value = (s.value + 1) % len(s.parts)
+			s.mu.Unlock()
 		}
 	}
 }
 
 func (s *Spinner) Stop() {
+	s.mu.Lock()
 	if s.stopped.IsZero() {
 		s.stopped = time.Now()
 	}
+	s.mu.Unlock()
+
+	s.stopOnce.Do(func() { close(s.done) })
 }