macOS

app/cmd/app/app.go

@@ -209,6 +209,9 @@ func main() {
 
 	st := &store.Store{}
 
+	// Initialize native settings with store
+	SetSettingsStore(st)
+
 	// Enable CORS in development mode
 	if devMode {
 		os.Setenv("OLLAMA_CORS", "1")
@@ -253,22 +256,27 @@ func main() {
 		done <- osrv.Run(octx)
 	}()
 
+	restartServer := func() {
+		ocancel()
+		<-done
+		octx, ocancel = context.WithCancel(ctx)
+		go func() {
+			done <- osrv.Run(octx)
+		}()
+	}
+
 	uiServer := ui.Server{
-		Token: token,
-		Restart: func() {
-			ocancel()
-			<-done
-			octx, ocancel = context.WithCancel(ctx)
-			go func() {
-				done <- osrv.Run(octx)
-			}()
-		},
+		Token:        token,
+		Restart:      restartServer,
 		Store:        st,
 		ToolRegistry: toolRegistry,
 		Dev:          devMode,
 		Logger:       slog.Default(),
 	}
 
+	// Set restart callback for native settings
+	SetRestartCallback(restartServer)
+
 	srv := &http.Server{
 		Handler: uiServer.Handler(),
 	}

app/cmd/app/app_darwin.m

@@ -1,5 +1,6 @@
 #import "app_darwin.h"
 #import "menu.h"
+#import "settings_darwin.h"
 #import "../../updater/updater_darwin.h"
 #import <AppKit/AppKit.h>
 #import <Cocoa/Cocoa.h>
@@ -252,7 +253,7 @@ bool firstTimeRun,startHidden; // Set in run before initialization
 }
 
 - (void)settingsUI {
-    [self uiRequest:@"/settings"];
+    openNativeSettings();
 }
 
 - (void)openUI {

app/cmd/app/settings_darwin.go

@@ -0,0 +1,438 @@
+//go:build darwin
+
+package main
+
+/*
+#cgo CFLAGS: -x objective-c
+#cgo LDFLAGS: -framework Cocoa
+
+#include <stdlib.h>
+#include "settings_darwin.h"
+*/
+import "C"
+
+import (
+	"context"
+	"crypto/ed25519"
+	"crypto/rand"
+	"encoding/json"
+	"encoding/pem"
+	"fmt"
+	"log/slog"
+	"net/http"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strconv"
+	"strings"
+	"time"
+	"unsafe"
+
+	"golang.org/x/crypto/ssh"
+
+	appauth "github.com/ollama/ollama/app/auth"
+	"github.com/ollama/ollama/app/store"
+	"github.com/ollama/ollama/auth"
+	"github.com/ollama/ollama/envconfig"
+)
+
+// settingsStore is a reference to the app's store for settings
+var settingsStore *store.Store
+
+// SetSettingsStore sets the store reference for settings callbacks
+func SetSettingsStore(s *store.Store) {
+	settingsStore = s
+}
+
+//export getSettingsExpose
+func getSettingsExpose() C.bool {
+	if settingsStore == nil {
+		return C.bool(false)
+	}
+	settings, err := settingsStore.Settings()
+	if err != nil {
+		slog.Error("failed to get settings", "error", err)
+		return C.bool(false)
+	}
+	return C.bool(settings.Expose)
+}
+
+//export setSettingsExpose
+func setSettingsExpose(expose C.bool) {
+	if settingsStore == nil {
+		return
+	}
+	settings, err := settingsStore.Settings()
+	if err != nil {
+		slog.Error("failed to get settings", "error", err)
+		return
+	}
+	settings.Expose = bool(expose)
+	if err := settingsStore.SetSettings(settings); err != nil {
+		slog.Error("failed to save settings", "error", err)
+	}
+}
+
+//export getSettingsBrowser
+func getSettingsBrowser() C.bool {
+	if settingsStore == nil {
+		return C.bool(false)
+	}
+	settings, err := settingsStore.Settings()
+	if err != nil {
+		slog.Error("failed to get settings", "error", err)
+		return C.bool(false)
+	}
+	return C.bool(settings.Browser)
+}
+
+//export setSettingsBrowser
+func setSettingsBrowser(browser C.bool) {
+	if settingsStore == nil {
+		return
+	}
+	settings, err := settingsStore.Settings()
+	if err != nil {
+		slog.Error("failed to get settings", "error", err)
+		return
+	}
+	settings.Browser = bool(browser)
+	if err := settingsStore.SetSettings(settings); err != nil {
+		slog.Error("failed to save settings", "error", err)
+	}
+}
+
+//export getSettingsModels
+func getSettingsModels() *C.char {
+	if settingsStore == nil {
+		return C.CString(envconfig.Models())
+	}
+	settings, err := settingsStore.Settings()
+	if err != nil {
+		slog.Error("failed to get settings", "error", err)
+		return C.CString(envconfig.Models())
+	}
+	if settings.Models == "" {
+		return C.CString(envconfig.Models())
+	}
+	return C.CString(settings.Models)
+}
+
+//export setSettingsModels
+func setSettingsModels(path *C.char) {
+	if settingsStore == nil {
+		return
+	}
+	settings, err := settingsStore.Settings()
+	if err != nil {
+		slog.Error("failed to get settings", "error", err)
+		return
+	}
+	settings.Models = C.GoString(path)
+	if err := settingsStore.SetSettings(settings); err != nil {
+		slog.Error("failed to save settings", "error", err)
+	}
+}
+
+//export getSettingsContextLength
+func getSettingsContextLength() C.int {
+	if settingsStore == nil {
+		return C.int(4096)
+	}
+	settings, err := settingsStore.Settings()
+	if err != nil {
+		slog.Error("failed to get settings", "error", err)
+		return C.int(4096)
+	}
+	if settings.ContextLength <= 0 {
+		return C.int(4096)
+	}
+	return C.int(settings.ContextLength)
+}
+
+//export setSettingsContextLength
+func setSettingsContextLength(length C.int) {
+	if settingsStore == nil {
+		return
+	}
+	settings, err := settingsStore.Settings()
+	if err != nil {
+		slog.Error("failed to get settings", "error", err)
+		return
+	}
+	settings.ContextLength = int(length)
+	if err := settingsStore.SetSettings(settings); err != nil {
+		slog.Error("failed to save settings", "error", err)
+	}
+}
+
+// restartCallback is set by the app to restart the ollama server
+var restartCallback func()
+
+// SetRestartCallback sets the function to call when settings change requires a restart
+func SetRestartCallback(cb func()) {
+	restartCallback = cb
+}
+
+//export restartOllamaServer
+func restartOllamaServer() {
+	if restartCallback != nil {
+		slog.Info("restarting ollama server due to settings change")
+		go restartCallback()
+	}
+}
+
+// hasOllamaKey checks if the user has an Ollama key file
+func hasOllamaKey() bool {
+	home, err := os.UserHomeDir()
+	if err != nil {
+		return false
+	}
+	keyPath := filepath.Join(home, ".ollama", "id_ed25519")
+	_, err = os.Stat(keyPath)
+	return err == nil
+}
+
+// ensureKeypair generates a new keypair if one doesn't exist
+func ensureKeypair() error {
+	home, err := os.UserHomeDir()
+	if err != nil {
+		return err
+	}
+
+	privKeyPath := filepath.Join(home, ".ollama", "id_ed25519")
+
+	// Check if key already exists
+	if _, err := os.Stat(privKeyPath); err == nil {
+		return nil // Key exists
+	}
+
+	// Generate new keypair
+	slog.Info("generating new keypair for ollama account")
+
+	pubKey, privKey, err := ed25519.GenerateKey(rand.Reader)
+	if err != nil {
+		return fmt.Errorf("failed to generate key: %w", err)
+	}
+
+	// Marshal private key
+	privKeyBytes, err := ssh.MarshalPrivateKey(privKey, "")
+	if err != nil {
+		return fmt.Errorf("failed to marshal private key: %w", err)
+	}
+
+	// Ensure directory exists
+	if err := os.MkdirAll(filepath.Dir(privKeyPath), 0o755); err != nil {
+		return fmt.Errorf("failed to create directory: %w", err)
+	}
+
+	// Write private key
+	if err := os.WriteFile(privKeyPath, pem.EncodeToMemory(privKeyBytes), 0o600); err != nil {
+		return fmt.Errorf("failed to write private key: %w", err)
+	}
+
+	// Write public key
+	sshPubKey, err := ssh.NewPublicKey(pubKey)
+	if err != nil {
+		return fmt.Errorf("failed to create ssh public key: %w", err)
+	}
+	pubKeyBytes := ssh.MarshalAuthorizedKey(sshPubKey)
+	pubKeyPath := filepath.Join(home, ".ollama", "id_ed25519.pub")
+	if err := os.WriteFile(pubKeyPath, pubKeyBytes, 0o644); err != nil {
+		return fmt.Errorf("failed to write public key: %w", err)
+	}
+
+	slog.Info("keypair generated successfully")
+	return nil
+}
+
+// userResponse matches the API response from ollama.com/api/me
+type userResponse struct {
+	Name      string `json:"name"`
+	Email     string `json:"email"`
+	Plan      string `json:"plan"`
+	AvatarURL string `json:"avatarurl"`
+}
+
+// fetchUserFromAPI fetches user data from ollama.com using signed request
+func fetchUserFromAPI() (*userResponse, error) {
+	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+	defer cancel()
+
+	timestamp := strconv.FormatInt(time.Now().Unix(), 10)
+	signString := fmt.Sprintf("POST,/api/me?ts=%s", timestamp)
+	signature, err := auth.Sign(ctx, []byte(signString))
+	if err != nil {
+		return nil, fmt.Errorf("failed to sign request: %w", err)
+	}
+
+	endpoint := fmt.Sprintf("https://ollama.com/api/me?ts=%s", timestamp)
+	req, err := http.NewRequestWithContext(ctx, http.MethodPost, endpoint, nil)
+	if err != nil {
+		return nil, fmt.Errorf("failed to create request: %w", err)
+	}
+	req.Header.Set("Authorization", fmt.Sprintf("Bearer %s", signature))
+
+	client := &http.Client{Timeout: 10 * time.Second}
+	resp, err := client.Do(req)
+	if err != nil {
+		return nil, fmt.Errorf("failed to call ollama.com: %w", err)
+	}
+	defer resp.Body.Close()
+
+	if resp.StatusCode != http.StatusOK {
+		return nil, fmt.Errorf("unexpected status: %d", resp.StatusCode)
+	}
+
+	var user userResponse
+	if err := json.NewDecoder(resp.Body).Decode(&user); err != nil {
+		return nil, fmt.Errorf("failed to parse response: %w", err)
+	}
+
+	// Make avatar URL absolute
+	if user.AvatarURL != "" && !strings.HasPrefix(user.AvatarURL, "http") {
+		user.AvatarURL = "https://ollama.com/" + user.AvatarURL
+	}
+
+	// Cache the avatar URL
+	cachedAvatarURL = user.AvatarURL
+
+	// Cache the user data
+	if settingsStore != nil {
+		storeUser := store.User{
+			Name:  user.Name,
+			Email: user.Email,
+			Plan:  user.Plan,
+		}
+		if err := settingsStore.SetUser(storeUser); err != nil {
+			slog.Warn("failed to cache user", "error", err)
+		}
+	}
+
+	return &user, nil
+}
+
+//export getAccountName
+func getAccountName() *C.char {
+	// Only return cached data - never block on network
+	if settingsStore == nil {
+		return C.CString("")
+	}
+	user, err := settingsStore.User()
+	if err != nil || user == nil {
+		return C.CString("")
+	}
+	return C.CString(user.Name)
+}
+
+// cachedAvatarURL stores the avatar URL from the last API fetch
+var cachedAvatarURL string
+
+//export getAccountAvatarURL
+func getAccountAvatarURL() *C.char {
+	return C.CString(cachedAvatarURL)
+}
+
+//export getAccountEmail
+func getAccountEmail() *C.char {
+	if settingsStore != nil {
+		user, err := settingsStore.User()
+		if err == nil && user != nil {
+			return C.CString(user.Email)
+		}
+	}
+	return C.CString("")
+}
+
+//export getAccountPlan
+func getAccountPlan() *C.char {
+	if settingsStore != nil {
+		user, err := settingsStore.User()
+		if err == nil && user != nil {
+			return C.CString(user.Plan)
+		}
+	}
+	return C.CString("")
+}
+
+//export signOutAccount
+func signOutAccount() {
+	if settingsStore != nil {
+		if err := settingsStore.ClearUser(); err != nil {
+			slog.Error("failed to clear user", "error", err)
+		}
+	}
+
+	// Also remove the key file
+	home, err := os.UserHomeDir()
+	if err != nil {
+		slog.Error("failed to get home dir", "error", err)
+		return
+	}
+	keyPath := filepath.Join(home, ".ollama", "id_ed25519")
+	if err := os.Remove(keyPath); err != nil && !os.IsNotExist(err) {
+		slog.Error("failed to remove key file", "error", err)
+	}
+}
+
+//export openConnectUrl
+func openConnectUrl() {
+	// Ensure keypair exists (generate if needed)
+	if err := ensureKeypair(); err != nil {
+		slog.Error("failed to ensure keypair", "error", err)
+		// Fallback to basic connect page
+		cmd := exec.Command("open", "https://ollama.com/connect")
+		cmd.Start()
+		return
+	}
+
+	// Build connect URL with public key
+	connectURL, err := appauth.BuildConnectURL("https://ollama.com")
+	if err != nil {
+		slog.Error("failed to build connect URL", "error", err)
+		// Fallback to basic connect page
+		connectURL = "https://ollama.com/connect"
+	}
+
+	cmd := exec.Command("open", connectURL)
+	if err := cmd.Start(); err != nil {
+		slog.Error("failed to open connect URL", "error", err)
+	}
+}
+
+//export refreshAccountFromAPI
+func refreshAccountFromAPI() {
+	if !hasOllamaKey() {
+		return
+	}
+	_, err := fetchUserFromAPI()
+	if err != nil {
+		slog.Debug("failed to refresh account", "error", err)
+	}
+}
+
+//export prefetchAccountData
+func prefetchAccountData() {
+	// Run in background goroutine to not block app startup
+	go func() {
+		if !hasOllamaKey() {
+			return
+		}
+		_, err := fetchUserFromAPI()
+		if err != nil {
+			slog.Debug("failed to prefetch account data", "error", err)
+		} else {
+			slog.Debug("prefetched account data successfully")
+		}
+	}()
+}
+
+// OpenNativeSettings opens the native settings window
+func OpenNativeSettings() {
+	C.openNativeSettings()
+}
+
+// Ensure the CString is freed (caller must free)
+func freeCString(s *C.char) {
+	C.free(unsafe.Pointer(s))
+}

app/cmd/app/settings_darwin.h

@@ -0,0 +1,38 @@
+#import <Cocoa/Cocoa.h>
+
+@interface SettingsWindowController : NSWindowController <NSWindowDelegate>
+
+// General tab
+@property(nonatomic, strong) NSButton *exposeCheckbox;
+@property(nonatomic, strong) NSButton *browserCheckbox;
+@property(nonatomic, strong) NSSlider *contextLengthSlider;
+
+// Models tab
+@property(nonatomic, strong) NSPathControl *modelsPathControl;
+@property(nonatomic, strong) NSButton *modelsPathButton;
+
+// Account tab
+@property(nonatomic, strong) NSView *avatarView;
+@property(nonatomic, strong) NSTextField *avatarInitialLabel;
+@property(nonatomic, strong) NSImageView *avatarImageView;
+@property(nonatomic, strong) NSTextField *accountNameLabel;
+@property(nonatomic, strong) NSTextField *accountEmailLabel;
+@property(nonatomic, strong) NSButton *manageButton;
+@property(nonatomic, strong) NSButton *signOutButton;
+@property(nonatomic, strong) NSButton *signInButton;
+@property(nonatomic, strong) NSView *signedInContainer;
+@property(nonatomic, strong) NSView *signedOutContainer;
+
+// Plan section
+@property(nonatomic, strong) NSView *planContainer;
+@property(nonatomic, strong) NSTextField *planNameLabel;
+@property(nonatomic, strong) NSButton *upgradeButton;
+@property(nonatomic, strong) NSButton *viewUsageButton;
+
++ (instancetype)sharedController;
+- (void)showSettings;
+
+@end
+
+// Go callbacks for settings
+void openNativeSettings(void);

app/cmd/app/settings_windows.go

@@ -0,0 +1,16 @@
+//go:build windows
+
+package main
+
+import "github.com/ollama/ollama/app/store"
+
+// SetSettingsStore sets the store reference for settings callbacks (stub for Windows)
+func SetSettingsStore(s *store.Store) {
+	// TODO: Implement Windows native settings
+}
+
+// SetRestartCallback sets the function to call when settings change requires a restart (stub for Windows)
+func SetRestartCallback(cb func()) {
+	// TODO: Implement Windows native settings
+}
+

server/download.go

@@ -2,11 +2,9 @@ package server
 
 import (
 	"context"
-	"crypto/sha256"
 	"encoding/json"
 	"errors"
 	"fmt"
-	"hash"
 	"io"
 	"log/slog"
 	"math"
@@ -33,45 +31,9 @@ const maxRetries = 6
 var (
 	errMaxRetriesExceeded   = errors.New("max retries exceeded")
 	errPartStalled          = errors.New("part stalled")
-	errPartSlow             = errors.New("part slow, racing")
 	errMaxRedirectsExceeded = errors.New("maximum redirects exceeded (10) for directURL")
 )
 
-// speedTracker tracks download speeds and computes rolling median.
-type speedTracker struct {
-	mu     sync.Mutex
-	speeds []float64 // bytes per second
-}
-
-func (s *speedTracker) Record(bytesPerSec float64) {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	s.speeds = append(s.speeds, bytesPerSec)
-	// Keep last 100 samples
-	if len(s.speeds) > 100 {
-		s.speeds = s.speeds[1:]
-	}
-}
-
-func (s *speedTracker) Median() float64 {
-	s.mu.Lock()
-	defer s.mu.Unlock()
-	if len(s.speeds) < 3 {
-		return 0 // not enough data
-	}
-	// Simple median: sort a copy and take middle
-	sorted := make([]float64, len(s.speeds))
-	copy(sorted, s.speeds)
-	for i := range sorted {
-		for j := i + 1; j < len(sorted); j++ {
-			if sorted[j] < sorted[i] {
-				sorted[i], sorted[j] = sorted[j], sorted[i]
-			}
-		}
-	}
-	return sorted[len(sorted)/2]
-}
-
 var blobDownloadManager sync.Map
 
 type blobDownload struct {
@@ -132,127 +94,26 @@ func (p *blobDownloadPart) UnmarshalJSON(b []byte) error {
 	return nil
 }
 
-var (
-	downloadPartSize    = int64(envInt("OLLAMA_DOWNLOAD_PART_SIZE", 64)) * format.MegaByte
-	downloadConcurrency = envInt("OLLAMA_DOWNLOAD_CONCURRENCY", 48)
+const (
+	numDownloadParts          = 16
+	minDownloadPartSize int64 = 100 * format.MegaByte
+	maxDownloadPartSize int64 = 1000 * format.MegaByte
 )
 
-func envInt(key string, defaultVal int) int {
-	if s := os.Getenv(key); s != "" {
-		if v, err := strconv.Atoi(s); err == nil {
-			return v
-		}
-	}
-	return defaultVal
-}
-
-// streamHasher reads a file sequentially and hashes it as chunks complete.
-// Memory usage: ~64KB (just the read buffer), regardless of file size or concurrency.
-// Works by reading from OS page cache - data just written is still in RAM.
-type streamHasher struct {
-	file   *os.File
-	hasher hash.Hash
-	parts  []*blobDownloadPart
-	total  int64 // total bytes to hash
-	hashed atomic.Int64
-
-	mu        sync.Mutex
-	cond      *sync.Cond
-	completed []bool
-	done      bool
-	err       error
-}
-
-func newStreamHasher(file *os.File, parts []*blobDownloadPart, total int64) *streamHasher {
-	h := &streamHasher{
-		file:      file,
-		hasher:    sha256.New(),
-		parts:     parts,
-		total:     total,
-		completed: make([]bool, len(parts)),
-	}
-	h.cond = sync.NewCond(&h.mu)
-	return h
-}
-
-// MarkComplete signals that a part has been written to disk.
-func (h *streamHasher) MarkComplete(partIndex int) {
-	h.mu.Lock()
-	h.completed[partIndex] = true
-	h.cond.Broadcast()
-	h.mu.Unlock()
-}
-
-// Run reads and hashes the file sequentially. Call in a goroutine.
-func (h *streamHasher) Run() {
-	buf := make([]byte, 64*1024) // 64KB read buffer
-	var offset int64
-
-	for i, part := range h.parts {
-		// Wait for this part to be written
-		h.mu.Lock()
-		for !h.completed[i] && !h.done {
-			h.cond.Wait()
-		}
-		if h.done {
-			h.mu.Unlock()
-			return
-		}
-		h.mu.Unlock()
-
-		// Read and hash this part (from page cache)
-		remaining := part.Size
-		for remaining > 0 {
-			n := int64(len(buf))
-			if n > remaining {
-				n = remaining
-			}
-			nr, err := h.file.ReadAt(buf[:n], offset)
-			if err != nil && err != io.EOF {
-				h.mu.Lock()
-				h.err = err
-				h.mu.Unlock()
-				return
-			}
-			h.hasher.Write(buf[:nr])
-			offset += int64(nr)
-			remaining -= int64(nr)
-			h.hashed.Store(offset)
-		}
-	}
-}
-
-// Stop signals the hasher to exit early.
-func (h *streamHasher) Stop() {
-	h.mu.Lock()
-	h.done = true
-	h.cond.Broadcast()
-	h.mu.Unlock()
-}
-
-// Hashed returns bytes hashed so far.
-func (h *streamHasher) Hashed() int64 {
-	return h.hashed.Load()
-}
-
-// Digest returns the computed hash.
-func (h *streamHasher) Digest() string {
-	return fmt.Sprintf("sha256:%x", h.hasher.Sum(nil))
-}
-
-// Err returns any error from hashing.
-func (h *streamHasher) Err() error {
-	h.mu.Lock()
-	defer h.mu.Unlock()
-	return h.err
-}
-
 func (p *blobDownloadPart) Name() string {
 	return strings.Join([]string{
 		p.blobDownload.Name, "partial", strconv.Itoa(p.N),
 	}, "-")
 }
 
+func (p *blobDownloadPart) StartsAt() int64 {
+	return p.Offset + p.Completed.Load()
+}
+
+func (p *blobDownloadPart) StopsAt() int64 {
+	return p.Offset + p.Size
+}
+
 func (p *blobDownloadPart) Write(b []byte) (n int, err error) {
 	n = len(b)
 	p.blobDownload.Completed.Add(int64(n))
@@ -290,7 +151,14 @@ func (b *blobDownload) Prepare(ctx context.Context, requestURL *url.URL, opts *r
 
 		b.Total, _ = strconv.ParseInt(resp.Header.Get("Content-Length"), 10, 64)
 
-		size := downloadPartSize
+		size := b.Total / numDownloadParts
+		switch {
+		case size < minDownloadPartSize:
+			size = minDownloadPartSize
+		case size > maxDownloadPartSize:
+			size = maxDownloadPartSize
+		}
+
 		var offset int64
 		for offset < b.Total {
 			if offset+size > b.Total {
@@ -352,6 +220,9 @@ func (b *blobDownload) run(ctx context.Context, requestURL *url.URL, opts *regis
 		return err
 	}
 	defer file.Close()
+	setSparse(file)
+
+	_ = file.Truncate(b.Total)
 
 	directURL, err := func() (*url.URL, error) {
 		ctx, cancel := context.WithTimeout(ctx, 30*time.Second)
@@ -399,106 +270,44 @@ func (b *blobDownload) run(ctx context.Context, requestURL *url.URL, opts *regis
 		return err
 	}
 
-	// Download chunks to disk, hash by reading from page cache.
-	// Memory: ~64KB (hasher read buffer only), regardless of concurrency.
-	// The hasher follows behind the downloaders, reading recently-written
-	// data from OS page cache (RAM) rather than disk.
-	sh := newStreamHasher(file, b.Parts, b.Total)
-	tracker := &speedTracker{}
-
-	// Start hasher goroutine
-	hashDone := make(chan struct{})
-	go func() {
-		sh.Run()
-		close(hashDone)
-	}()
-
-	// Log progress periodically
-	// Page cache warning: if spread > 4GB, hasher may hit disk instead of RAM
-	const pageCacheWarningBytes = 4 << 30 // 4GB
-	progressDone := make(chan struct{})
-	go func() {
-		ticker := time.NewTicker(time.Second)
-		defer ticker.Stop()
-		for {
-			select {
-			case <-ticker.C:
-				downloaded := b.Completed.Load()
-				hashed := sh.Hashed()
-				dlPct := int(downloaded * 100 / b.Total)
-				hPct := int(hashed * 100 / b.Total)
-				spread := dlPct - hPct
-				spreadBytes := downloaded - hashed
-
-				slog.Debug(fmt.Sprintf("progress: downloaded %d%% | hashed %d%% | spread %d%%", dlPct, hPct, spread))
-				if spreadBytes > pageCacheWarningBytes {
-					slog.Debug("page cache pressure", "ahead", fmt.Sprintf("%.1fGB", float64(spreadBytes)/(1<<30)))
-				}
-			case <-progressDone:
-				return
-			}
-		}
-	}()
-
 	g, inner := errgroup.WithContext(ctx)
-	g.SetLimit(downloadConcurrency)
+	g.SetLimit(numDownloadParts)
 	for i := range b.Parts {
 		part := b.Parts[i]
 		if part.Completed.Load() == part.Size {
-			sh.MarkComplete(part.N)
 			continue
 		}
 
 		g.Go(func() error {
 			var err error
-			var slowRetries int
 			for try := 0; try < maxRetries; try++ {
-				// After 3 slow retries, stop checking slowness and let it complete
-				skipSlowCheck := slowRetries >= 3
-				err = b.downloadChunkToDisk(inner, directURL, file, part, tracker, skipSlowCheck)
+				w := io.NewOffsetWriter(file, part.StartsAt())
+				err = b.downloadChunk(inner, directURL, w, part)
 				switch {
 				case errors.Is(err, context.Canceled), errors.Is(err, syscall.ENOSPC):
+					// return immediately if the context is canceled or the device is out of space
 					return err
 				case errors.Is(err, errPartStalled):
 					try--
 					continue
-				case errors.Is(err, errPartSlow):
-					// Kill slow request, retry immediately (stays within concurrency limit)
-					slowRetries++
-					try--
-					continue
 				case err != nil:
 					sleep := time.Second * time.Duration(math.Pow(2, float64(try)))
 					slog.Info(fmt.Sprintf("%s part %d attempt %d failed: %v, retrying in %s", b.Digest[7:19], part.N, try, err, sleep))
 					time.Sleep(sleep)
 					continue
 				default:
-					sh.MarkComplete(part.N)
 					return nil
 				}
 			}
+
 			return fmt.Errorf("%w: %w", errMaxRetriesExceeded, err)
 		})
 	}
 
 	if err := g.Wait(); err != nil {
-		close(progressDone)
-		sh.Stop()
 		return err
 	}
 
-	// Wait for hasher to finish
-	<-hashDone
-	close(progressDone)
-	if err := sh.Err(); err != nil {
-		return err
-	}
-
-	// Verify hash
-	if computed := sh.Digest(); computed != b.Digest {
-		return fmt.Errorf("digest mismatch: got %s, want %s", computed, b.Digest)
-	}
-
 	// explicitly close the file so we can rename it
 	if err := file.Close(); err != nil {
 		return err
@@ -517,69 +326,38 @@ func (b *blobDownload) run(ctx context.Context, requestURL *url.URL, opts *regis
 	return nil
 }
 
-// downloadChunkToDisk streams a part directly to disk at its offset.
-// Memory: ~32KB (read buffer only).
-// If skipSlowCheck is true, don't flag slow parts (used after repeated slow retries).
-func (b *blobDownload) downloadChunkToDisk(ctx context.Context, requestURL *url.URL, file *os.File, part *blobDownloadPart, tracker *speedTracker, skipSlowCheck bool) error {
+func (b *blobDownload) downloadChunk(ctx context.Context, requestURL *url.URL, w io.Writer, part *blobDownloadPart) error {
 	g, ctx := errgroup.WithContext(ctx)
-	startTime := time.Now()
-	var bytesAtLastCheck atomic.Int64
-
 	g.Go(func() error {
 		req, err := http.NewRequestWithContext(ctx, http.MethodGet, requestURL.String(), nil)
 		if err != nil {
 			return err
 		}
-		req.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", part.Offset, part.Offset+part.Size-1))
+		req.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", part.StartsAt(), part.StopsAt()-1))
 		resp, err := http.DefaultClient.Do(req)
 		if err != nil {
 			return err
 		}
 		defer resp.Body.Close()
 
-		w := io.NewOffsetWriter(file, part.Offset)
-		buf := make([]byte, 32*1024)
-
-		var written int64
-		for written < part.Size {
-			n, err := resp.Body.Read(buf)
-			if n > 0 {
-				if _, werr := w.Write(buf[:n]); werr != nil {
-					return werr
-				}
-				written += int64(n)
-				b.Completed.Add(int64(n))
-				bytesAtLastCheck.Store(written)
-
-				part.lastUpdatedMu.Lock()
-				part.lastUpdated = time.Now()
-				part.lastUpdatedMu.Unlock()
-			}
-			if err == io.EOF {
-				break
-			}
-			if err != nil {
-				b.Completed.Add(-written)
-				return err
-			}
+		n, err := io.CopyN(w, io.TeeReader(resp.Body, part), part.Size-part.Completed.Load())
+		if err != nil && !errors.Is(err, context.Canceled) && !errors.Is(err, io.ErrUnexpectedEOF) {
+			// rollback progress
+			b.Completed.Add(-n)
+			return err
 		}
 
-		// Record speed for this part
-		elapsed := time.Since(startTime).Seconds()
-		if elapsed > 0 {
-			tracker.Record(float64(part.Size) / elapsed)
+		part.Completed.Add(n)
+		if err := b.writePart(part.Name(), part); err != nil {
+			return err
 		}
 
-		part.Completed.Store(part.Size)
-		return b.writePart(part.Name(), part)
+		// return nil or context.Canceled or UnexpectedEOF (resumable)
+		return err
 	})
 
 	g.Go(func() error {
 		ticker := time.NewTicker(time.Second)
-		defer ticker.Stop()
-		var lastBytes int64
-		checksWithoutProgress := 0
-
 		for {
 			select {
 			case <-ticker.C:
@@ -587,47 +365,19 @@ func (b *blobDownload) downloadChunkToDisk(ctx context.Context, requestURL *url.
 					return nil
 				}
 
-				currentBytes := bytesAtLastCheck.Load()
-
-				// Check for complete stall (30 seconds no progress)
 				part.lastUpdatedMu.Lock()
 				lastUpdated := part.lastUpdated
 				part.lastUpdatedMu.Unlock()
+
 				if !lastUpdated.IsZero() && time.Since(lastUpdated) > 30*time.Second {
-					slog.Info(fmt.Sprintf("%s part %d stalled; retrying", b.Digest[7:19], part.N))
+					const msg = "%s part %d stalled; retrying. If this persists, press ctrl-c to exit, then 'ollama pull' to find a faster connection."
+					slog.Info(fmt.Sprintf(msg, b.Digest[7:19], part.N))
+					// reset last updated
 					part.lastUpdatedMu.Lock()
 					part.lastUpdated = time.Time{}
 					part.lastUpdatedMu.Unlock()
 					return errPartStalled
 				}
-
-				// Check for slow speed after 5+ seconds (only for multi-part downloads)
-				// Skip if we've already retried for slowness too many times
-				elapsed := time.Since(startTime).Seconds()
-				if !skipSlowCheck && elapsed >= 5 && currentBytes > 0 && len(b.Parts) > 1 {
-					currentSpeed := float64(currentBytes) / elapsed
-					median := tracker.Median()
-
-					// If we're below 10% of median speed, flag as slow
-					if median > 0 && currentSpeed < median*0.1 {
-						slog.Info(fmt.Sprintf("%s part %d slow (%.0f KB/s vs median %.0f KB/s); retrying",
-							b.Digest[7:19], part.N, currentSpeed/1024, median/1024))
-						return errPartSlow
-					}
-				}
-
-				// Also check if speed dropped significantly mid-download
-				if currentBytes == lastBytes {
-					checksWithoutProgress++
-					if checksWithoutProgress >= 10 {
-						slog.Info(fmt.Sprintf("%s part %d no progress for 10s; retrying", b.Digest[7:19], part.N))
-						return errPartStalled
-					}
-				} else {
-					checksWithoutProgress = 0
-				}
-				lastBytes = currentBytes
-
 			case <-ctx.Done():
 				return ctx.Err()
 			}

server/download_test.go

@@ -1,319 +0,0 @@
-package server
-
-import (
-	"crypto/rand"
-	"crypto/sha256"
-	"fmt"
-	"os"
-	"sync"
-	"testing"
-)
-
-func TestSpeedTracker_Median(t *testing.T) {
-	s := &speedTracker{}
-
-	// Less than 3 samples returns 0
-	s.Record(100)
-	s.Record(200)
-	if got := s.Median(); got != 0 {
-		t.Errorf("expected 0 with < 3 samples, got %f", got)
-	}
-
-	// With 3+ samples, returns median
-	s.Record(300)
-	// Samples: [100, 200, 300] -> median = 200
-	if got := s.Median(); got != 200 {
-		t.Errorf("expected median 200, got %f", got)
-	}
-
-	// Add more samples
-	s.Record(50)
-	s.Record(250)
-	// Samples: [100, 200, 300, 50, 250] sorted = [50, 100, 200, 250, 300] -> median = 200
-	if got := s.Median(); got != 200 {
-		t.Errorf("expected median 200, got %f", got)
-	}
-}
-
-func TestSpeedTracker_RollingWindow(t *testing.T) {
-	s := &speedTracker{}
-
-	// Add 105 samples (should keep only last 100)
-	for i := 0; i < 105; i++ {
-		s.Record(float64(i))
-	}
-
-	s.mu.Lock()
-	if len(s.speeds) != 100 {
-		t.Errorf("expected 100 samples, got %d", len(s.speeds))
-	}
-	// First sample should be 5 (0-4 were dropped)
-	if s.speeds[0] != 5 {
-		t.Errorf("expected first sample to be 5, got %f", s.speeds[0])
-	}
-	s.mu.Unlock()
-}
-
-func TestSpeedTracker_Concurrent(t *testing.T) {
-	s := &speedTracker{}
-
-	var wg sync.WaitGroup
-	for i := 0; i < 100; i++ {
-		wg.Add(1)
-		go func(v int) {
-			defer wg.Done()
-			s.Record(float64(v))
-			s.Median() // concurrent read
-		}(i)
-	}
-	wg.Wait()
-
-	// Should not panic, and should have reasonable state
-	s.mu.Lock()
-	if len(s.speeds) == 0 || len(s.speeds) > 100 {
-		t.Errorf("unexpected speeds length: %d", len(s.speeds))
-	}
-	s.mu.Unlock()
-}
-
-func TestStreamHasher_Sequential(t *testing.T) {
-	// Create temp file
-	f, err := os.CreateTemp("", "streamhasher_test")
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer os.Remove(f.Name())
-	defer f.Close()
-
-	// Write test data
-	data := []byte("hello world, this is a test of the stream hasher")
-	if _, err := f.Write(data); err != nil {
-		t.Fatal(err)
-	}
-
-	// Create parts
-	parts := []*blobDownloadPart{
-		{Offset: 0, Size: int64(len(data))},
-	}
-
-	sh := newStreamHasher(f, parts, int64(len(data)))
-
-	// Mark complete and run
-	sh.MarkComplete(0)
-
-	done := make(chan struct{})
-	go func() {
-		sh.Run()
-		close(done)
-	}()
-	<-done
-
-	// Verify digest
-	expected := fmt.Sprintf("sha256:%x", sha256.Sum256(data))
-	if got := sh.Digest(); got != expected {
-		t.Errorf("digest mismatch: got %s, want %s", got, expected)
-	}
-
-	if err := sh.Err(); err != nil {
-		t.Errorf("unexpected error: %v", err)
-	}
-}
-
-func TestStreamHasher_OutOfOrderCompletion(t *testing.T) {
-	// Create temp file
-	f, err := os.CreateTemp("", "streamhasher_test")
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer os.Remove(f.Name())
-	defer f.Close()
-
-	// Write test data (3 parts of 10 bytes each)
-	data := []byte("0123456789ABCDEFGHIJabcdefghij")
-	if _, err := f.Write(data); err != nil {
-		t.Fatal(err)
-	}
-
-	// Create 3 parts
-	parts := []*blobDownloadPart{
-		{N: 0, Offset: 0, Size: 10},
-		{N: 1, Offset: 10, Size: 10},
-		{N: 2, Offset: 20, Size: 10},
-	}
-
-	sh := newStreamHasher(f, parts, int64(len(data)))
-
-	done := make(chan struct{})
-	go func() {
-		sh.Run()
-		close(done)
-	}()
-
-	// Mark parts complete out of order: 2, 0, 1
-	sh.MarkComplete(2)
-	sh.MarkComplete(0) // This should trigger hashing of part 0
-	sh.MarkComplete(1) // This should trigger hashing of parts 1 and 2
-
-	<-done
-
-	// Verify digest
-	expected := fmt.Sprintf("sha256:%x", sha256.Sum256(data))
-	if got := sh.Digest(); got != expected {
-		t.Errorf("digest mismatch: got %s, want %s", got, expected)
-	}
-}
-
-func TestStreamHasher_Stop(t *testing.T) {
-	// Create temp file
-	f, err := os.CreateTemp("", "streamhasher_test")
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer os.Remove(f.Name())
-	defer f.Close()
-
-	parts := []*blobDownloadPart{
-		{Offset: 0, Size: 100},
-	}
-
-	sh := newStreamHasher(f, parts, 100)
-
-	done := make(chan struct{})
-	go func() {
-		sh.Run()
-		close(done)
-	}()
-
-	// Stop without completing any parts
-	sh.Stop()
-	<-done
-
-	// Should exit cleanly without error
-	if err := sh.Err(); err != nil {
-		t.Errorf("unexpected error after Stop: %v", err)
-	}
-}
-
-func TestStreamHasher_HashedProgress(t *testing.T) {
-	// Create temp file with known data
-	f, err := os.CreateTemp("", "streamhasher_test")
-	if err != nil {
-		t.Fatal(err)
-	}
-	defer os.Remove(f.Name())
-	defer f.Close()
-
-	data := make([]byte, 1000)
-	rand.Read(data)
-	if _, err := f.Write(data); err != nil {
-		t.Fatal(err)
-	}
-
-	parts := []*blobDownloadPart{
-		{N: 0, Offset: 0, Size: 500},
-		{N: 1, Offset: 500, Size: 500},
-	}
-
-	sh := newStreamHasher(f, parts, 1000)
-
-	// Initially no progress
-	if got := sh.Hashed(); got != 0 {
-		t.Errorf("expected 0 hashed initially, got %d", got)
-	}
-
-	done := make(chan struct{})
-	go func() {
-		sh.Run()
-		close(done)
-	}()
-
-	// Complete part 0
-	sh.MarkComplete(0)
-
-	// Give hasher time to process
-	for i := 0; i < 100; i++ {
-		if sh.Hashed() >= 500 {
-			break
-		}
-	}
-
-	// Complete part 1
-	sh.MarkComplete(1)
-	<-done
-
-	if got := sh.Hashed(); got != 1000 {
-		t.Errorf("expected 1000 hashed, got %d", got)
-	}
-}
-
-func BenchmarkSpeedTracker_Record(b *testing.B) {
-	s := &speedTracker{}
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		s.Record(float64(i))
-	}
-}
-
-func BenchmarkSpeedTracker_Median(b *testing.B) {
-	s := &speedTracker{}
-	// Pre-populate with 100 samples
-	for i := 0; i < 100; i++ {
-		s.Record(float64(i))
-	}
-	b.ResetTimer()
-	for i := 0; i < b.N; i++ {
-		s.Median()
-	}
-}
-
-func BenchmarkStreamHasher(b *testing.B) {
-	// Create temp file with test data
-	f, err := os.CreateTemp("", "streamhasher_bench")
-	if err != nil {
-		b.Fatal(err)
-	}
-	defer os.Remove(f.Name())
-	defer f.Close()
-
-	size := 64 * 1024 * 1024 // 64MB
-	data := make([]byte, size)
-	rand.Read(data)
-	if _, err := f.Write(data); err != nil {
-		b.Fatal(err)
-	}
-
-	parts := []*blobDownloadPart{
-		{Offset: 0, Size: int64(size)},
-	}
-
-	b.SetBytes(int64(size))
-	b.ResetTimer()
-
-	for i := 0; i < b.N; i++ {
-		sh := newStreamHasher(f, parts, int64(size))
-		sh.MarkComplete(0)
-
-		done := make(chan struct{})
-		go func() {
-			sh.Run()
-			close(done)
-		}()
-		<-done
-	}
-}
-
-func BenchmarkHashThroughput(b *testing.B) {
-	// Baseline: raw SHA256 throughput on this machine
-	size := 256 * 1024 * 1024 // 256MB
-	data := make([]byte, size)
-	rand.Read(data)
-
-	b.SetBytes(int64(size))
-	b.ResetTimer()
-
-	for i := 0; i < b.N; i++ {
-		h := sha256.New()
-		h.Write(data)
-		h.Sum(nil)
-	}
-}

server/images.go

@@ -620,8 +620,9 @@ func PullModel(ctx context.Context, name string, regOpts *registryOptions, fn fu
 		layers = append(layers, manifest.Config)
 	}
 
+	skipVerify := make(map[string]bool)
 	for _, layer := range layers {
-		_, err := downloadBlob(ctx, downloadOpts{
+		cacheHit, err := downloadBlob(ctx, downloadOpts{
 			mp:      mp,
 			digest:  layer.Digest,
 			regOpts: regOpts,
@@ -630,12 +631,31 @@ func PullModel(ctx context.Context, name string, regOpts *registryOptions, fn fu
 		if err != nil {
 			return err
 		}
+		skipVerify[layer.Digest] = cacheHit
 		delete(deleteMap, layer.Digest)
 	}
 	delete(deleteMap, manifest.Config.Digest)
 
-	// Note: Digest verification now happens inline during download in blobDownload.run()
-	// via the orderedWriter, so no separate verification pass is needed.
+	fn(api.ProgressResponse{Status: "verifying sha256 digest"})
+	for _, layer := range layers {
+		if skipVerify[layer.Digest] {
+			continue
+		}
+		if err := verifyBlob(layer.Digest); err != nil {
+			if errors.Is(err, errDigestMismatch) {
+				// something went wrong, delete the blob
+				fp, err := GetBlobsPath(layer.Digest)
+				if err != nil {
+					return err
+				}
+				if err := os.Remove(fp); err != nil {
+					// log this, but return the original error
+					slog.Info(fmt.Sprintf("couldn't remove file with digest mismatch '%s': %v", fp, err))
+				}
+			}
+			return err
+		}
+	}
 
 	fn(api.ProgressResponse{Status: "writing manifest"})
 

server/internal/cache/blob/cache.go

@@ -10,6 +10,7 @@ import (
 	"hash"
 	"io"
 	"io/fs"
+	"iter"
 	"os"
 	"path/filepath"
 	"strings"
@@ -326,19 +327,21 @@ func (c *DiskCache) GetFile(d Digest) string {
 	return absJoin(c.dir, "blobs", filename)
 }
 
-// Links returns a slice of link names in lexical order.
+// Links returns a sequence of link names. The sequence is in lexical order.
 // Names are converted from their relative path form to their name form but are
 // not guaranteed to be valid. Callers should validate the names before using.
-func (c *DiskCache) Links() ([]string, error) {
-	paths, err := c.links()
-	if err != nil {
-		return nil, err
+func (c *DiskCache) Links() iter.Seq2[string, error] {
+	return func(yield func(string, error) bool) {
+		for path, err := range c.links() {
+			if err != nil {
+				yield("", err)
+				return
+			}
+			if !yield(pathToName(path), nil) {
+				return
+			}
+		}
 	}
-	names := make([]string, len(paths))
-	for i, path := range paths {
-		names[i] = pathToName(path)
-	}
-	return names, nil
 }
 
 // pathToName converts a path to a name. It is the inverse of nameToPath. The
@@ -369,11 +372,10 @@ func (c *DiskCache) manifestPath(name string) (string, error) {
 	}
 
 	maybe := filepath.Join("manifests", np)
-	paths, err := c.links()
-	if err != nil {
-		return "", err
-	}
-	for _, l := range paths {
+	for l, err := range c.links() {
+		if err != nil {
+			return "", err
+		}
 		if strings.EqualFold(maybe, l) {
 			return filepath.Join(c.dir, l), nil
 		}
@@ -381,10 +383,22 @@ func (c *DiskCache) manifestPath(name string) (string, error) {
 	return filepath.Join(c.dir, maybe), nil
 }
 
-// links returns a slice of link paths in the cache in lexical order.
-func (c *DiskCache) links() ([]string, error) {
-	fsys := os.DirFS(c.dir)
-	return fs.Glob(fsys, "manifests/*/*/*/*")
+// links returns a sequence of links in the cache in lexical order.
+func (c *DiskCache) links() iter.Seq2[string, error] {
+	// TODO(bmizerany): reuse empty dirnames if exist
+	return func(yield func(string, error) bool) {
+		fsys := os.DirFS(c.dir)
+		manifests, err := fs.Glob(fsys, "manifests/*/*/*/*")
+		if err != nil {
+			yield("", err)
+			return
+		}
+		for _, manifest := range manifests {
+			if !yield(manifest, nil) {
+				return
+			}
+		}
+	}
 }
 
 type checkWriter struct {

server/internal/cache/blob/cache_test.go

@@ -466,9 +466,12 @@ func testManifestNameReuse(t *testing.T) {
 		t.Fatalf("g = %v, want %v", g, w)
 	}
 
-	got, err := c.links()
-	if err != nil {
-		t.Fatal(err)
+	var got []string
+	for l, err := range c.links() {
+		if err != nil {
+			t.Fatal(err)
+		}
+		got = append(got, l)
 	}
 	want := []string{"manifests/h/n/m/t"}
 	if !slices.Equal(got, want) {
@@ -484,9 +487,12 @@ func testManifestNameReuse(t *testing.T) {
 	err = c.Link("h/n/m:T", d1)
 	check(err)
 
-	got, err = c.links()
-	if err != nil {
-		t.Fatal(err)
+	got = got[:0]
+	for l, err := range c.links() {
+		if err != nil {
+			t.Fatal(err)
+		}
+		got = append(got, l)
 	}
 
 	// we should have only one link that is same case as the last link
@@ -548,9 +554,12 @@ func TestNames(t *testing.T) {
 	check(c.Link("h/n/m:t", mkdigest("1")))
 	check(c.Link("h/n/m:u", mkdigest("2")))
 
-	got, err := c.Links()
-	if err != nil {
-		t.Fatal(err)
+	var got []string
+	for l, err := range c.Links() {
+		if err != nil {
+			t.Fatal(err)
+		}
+		got = append(got, l)
 	}
 	want := []string{"h/n/m:t", "h/n/m:u"}
 	if !slices.Equal(got, want) {

server/internal/client/ollama/registry.go

@@ -19,6 +19,7 @@ import (
 	"fmt"
 	"io"
 	"io/fs"
+	"iter"
 	"log/slog"
 	"net/http"
 	"os"
@@ -545,7 +546,18 @@ func (r *Registry) Pull(ctx context.Context, name string) error {
 				})
 			}()
 
-			err = r.chunksums(ctx, name, l, func(cs chunksum) bool {
+			for cs, err := range r.chunksums(ctx, name, l) {
+				if err != nil {
+					// Note the chunksum stream
+					// interruption, but do not cancel
+					// in-flight downloads. We can still
+					// make progress on them. Once they are
+					// done, ErrIncomplete will be returned
+					// below.
+					update(0, err)
+					break
+				}
+
 				cacheKey := fmt.Sprintf(
 					"v1 pull chunksum %s %s %d-%d",
 					l.Digest,
@@ -557,7 +569,7 @@ func (r *Registry) Pull(ctx context.Context, name string) error {
 				_, err := c.Get(cacheKeyDigest)
 				if err == nil {
 					update(cs.Chunk.Size(), ErrCached)
-					return true // continue
+					continue
 				}
 
 				wg.Add(1)
@@ -608,13 +620,6 @@ func (r *Registry) Pull(ctx context.Context, name string) error {
 					// Record the downloading of this chunk.
 					return blob.PutBytes(c, cacheKeyDigest, cacheKey)
 				})
-				return true // continue processing chunks
-			})
-			if err != nil {
-				// Note the chunksum stream interruption, but do not cancel
-				// in-flight downloads. We can still make progress on them.
-				// Once they are done, ErrIncomplete will be returned below.
-				update(0, err)
 			}
 
 			return nil
@@ -669,6 +674,19 @@ func (m *Manifest) Layer(d blob.Digest) *Layer {
 	return nil
 }
 
+func (m *Manifest) All() iter.Seq[*Layer] {
+	return func(yield func(*Layer) bool) {
+		if !yield(m.Config) {
+			return
+		}
+		for _, l := range m.Layers {
+			if !yield(l) {
+				return
+			}
+		}
+	}
+}
+
 func (m *Manifest) Size() int64 {
 	var size int64
 	if m.Config != nil {
@@ -793,114 +811,125 @@ type chunksum struct {
 	Digest blob.Digest
 }
 
-// chunksums calls fn for each chunksum in the layer. If the layer is under the
-// chunking threshold, a single chunksum covering the entire layer is passed to fn.
-// If the layer is over the chunking threshold, chunksums are read from the chunksums endpoint.
-// Returns an error if the chunksum stream fails, or nil if all chunksums were processed.
-// If fn returns false, iteration stops early and chunksums returns nil.
-func (r *Registry) chunksums(ctx context.Context, name string, l *Layer, fn func(chunksum) bool) error {
-	scheme, n, _, err := r.parseNameExtended(name)
-	if err != nil {
-		return err
-	}
-
-	if l.Size < r.maxChunkingThreshold() {
-		// any layer under the threshold should be downloaded
-		// in one go.
-		cs := chunksum{
-			URL: fmt.Sprintf("%s://%s/v2/%s/%s/blobs/%s",
-				scheme,
-				n.Host(),
-				n.Namespace(),
-				n.Model(),
-				l.Digest,
-			),
-			Chunk:  blob.Chunk{Start: 0, End: l.Size - 1},
-			Digest: l.Digest,
-		}
-		fn(cs)
-		return nil
-	}
-
-	// The response is a sequence of chunksums.
-	//
-	// Chunksums are chunks of a larger blob that can be
-	// downloaded and verified independently.
-	//
-	// The chunksums endpoint is a GET request that returns a
-	// sequence of chunksums in the following format:
-	//
-	//     > GET /v2/<namespace>/<model>/chunksums/<digest>
-	//
-	//     < HTTP/1.1 200 OK
-	//     < Content-Location: <blobURL>
-	//     <
-	//     < <digest> <start>-<end>
-	//     < ...
-	//
-	// The <blobURL> is the URL to download the chunks from and
-	// each <digest> is the digest of the chunk, and <start>-<end>
-	// is the range the chunk in the blob.
-	//
-	// Ranges may be used directly in Range headers like
-	// "bytes=<start>-<end>".
-	//
-	// The chunksums returned are guaranteed to be contiguous and
-	// include all bytes of the layer. If the stream is cut short,
-	// clients should retry.
-
-	chunksumsURL := fmt.Sprintf("%s://%s/v2/%s/%s/chunksums/%s",
-		scheme,
-		n.Host(),
-		n.Namespace(),
-		n.Model(),
-		l.Digest,
-	)
-
-	req, err := r.newRequest(ctx, "GET", chunksumsURL, nil)
-	if err != nil {
-		return err
-	}
-	res, err := sendRequest(r.client(), req)
-	if err != nil {
-		return err
-	}
-	defer res.Body.Close()
-	if res.StatusCode != 200 {
-		return fmt.Errorf("chunksums: unexpected status code %d", res.StatusCode)
-	}
-	blobURL := res.Header.Get("Content-Location")
-
-	s := bufio.NewScanner(res.Body)
-	s.Split(bufio.ScanWords)
-	for {
-		if !s.Scan() {
-			return s.Err()
-		}
-		d, err := blob.ParseDigest(s.Bytes())
+// chunksums returns a sequence of chunksums for the given layer. If the layer is under the
+// chunking threshold, a single chunksum is returned that covers the entire layer. If the layer
+// is over the chunking threshold, the chunksums are read from the chunksums endpoint.
+func (r *Registry) chunksums(ctx context.Context, name string, l *Layer) iter.Seq2[chunksum, error] {
+	return func(yield func(chunksum, error) bool) {
+		scheme, n, _, err := r.parseNameExtended(name)
 		if err != nil {
-			return fmt.Errorf("invalid digest: %q", s.Bytes())
+			yield(chunksum{}, err)
+			return
 		}
 
-		if !s.Scan() {
-			err := s.Err()
-			if err == nil {
-				err = fmt.Errorf("missing chunk range for digest %s", d)
+		if l.Size < r.maxChunkingThreshold() {
+			// any layer under the threshold should be downloaded
+			// in one go.
+			cs := chunksum{
+				URL: fmt.Sprintf("%s://%s/v2/%s/%s/blobs/%s",
+					scheme,
+					n.Host(),
+					n.Namespace(),
+					n.Model(),
+					l.Digest,
+				),
+				Chunk:  blob.Chunk{Start: 0, End: l.Size - 1},
+				Digest: l.Digest,
 			}
-			return err
-		}
-		chunk, err := parseChunk(s.Bytes())
-		if err != nil {
-			return fmt.Errorf("invalid chunk range for digest %s: %q", d, s.Bytes())
+			yield(cs, nil)
+			return
 		}
 
-		cs := chunksum{
-			URL:    blobURL,
-			Chunk:  chunk,
-			Digest: d,
+		// The response is a sequence of chunksums.
+		//
+		// Chunksums are chunks of a larger blob that can be
+		// downloaded and verified independently.
+		//
+		// The chunksums endpoint is a GET request that returns a
+		// sequence of chunksums in the following format:
+		//
+		//     > GET /v2/<namespace>/<model>/chunksums/<digest>
+		//
+		//     < HTTP/1.1 200 OK
+		//     < Content-Location: <blobURL>
+		//     <
+		//     < <digest> <start>-<end>
+		//     < ...
+		//
+		// The <blobURL> is the URL to download the chunks from and
+		// each <digest> is the digest of the chunk, and <start>-<end>
+		// is the range the chunk in the blob.
+		//
+		// Ranges may be used directly in Range headers like
+		// "bytes=<start>-<end>".
+		//
+		// The chunksums returned are guaranteed to be contiguous and
+		// include all bytes of the layer. If the stream is cut short,
+		// clients should retry.
+
+		chunksumsURL := fmt.Sprintf("%s://%s/v2/%s/%s/chunksums/%s",
+			scheme,
+			n.Host(),
+			n.Namespace(),
+			n.Model(),
+			l.Digest,
+		)
+
+		req, err := r.newRequest(ctx, "GET", chunksumsURL, nil)
+		if err != nil {
+			yield(chunksum{}, err)
+			return
 		}
-		if !fn(cs) {
-			return nil
+		res, err := sendRequest(r.client(), req)
+		if err != nil {
+			yield(chunksum{}, err)
+			return
+		}
+		defer res.Body.Close()
+		if res.StatusCode != 200 {
+			err := fmt.Errorf("chunksums: unexpected status code %d", res.StatusCode)
+			yield(chunksum{}, err)
+			return
+		}
+		blobURL := res.Header.Get("Content-Location")
+
+		s := bufio.NewScanner(res.Body)
+		s.Split(bufio.ScanWords)
+		for {
+			if !s.Scan() {
+				if s.Err() != nil {
+					yield(chunksum{}, s.Err())
+				}
+				return
+			}
+			d, err := blob.ParseDigest(s.Bytes())
+			if err != nil {
+				yield(chunksum{}, fmt.Errorf("invalid digest: %q", s.Bytes()))
+				return
+			}
+
+			if !s.Scan() {
+				err := s.Err()
+				if err == nil {
+					err = fmt.Errorf("missing chunk range for digest %s", d)
+				}
+				yield(chunksum{}, err)
+				return
+			}
+			chunk, err := parseChunk(s.Bytes())
+			if err != nil {
+				yield(chunksum{}, fmt.Errorf("invalid chunk range for digest %s: %q", d, s.Bytes()))
+				return
+			}
+
+			cs := chunksum{
+				URL:    blobURL,
+				Chunk:  chunk,
+				Digest: d,
+			}
+			if !yield(cs, nil) {
+				return
+			}
 		}
 	}
 }
@@ -1147,8 +1176,8 @@ func splitExtended(s string) (scheme, name, digest string) {
 	return scheme, s, digest
 }
 
-// parseChunk parses a byte slice in the form "start-end" and returns the Chunk.
-func parseChunk(s []byte) (blob.Chunk, error) {
+// parseChunk parses a string in the form "start-end" and returns the Chunk.
+func parseChunk[S ~string | ~[]byte](s S) (blob.Chunk, error) {
 	startPart, endPart, found := strings.Cut(string(s), "-")
 	if !found {
 		return blob.Chunk{}, fmt.Errorf("chunks: invalid range %q: missing '-'", s)

server/internal/client/ollama/trace.go

@@ -27,20 +27,46 @@ type Trace struct {
 }
 
 func (t *Trace) update(l *Layer, n int64, err error) {
-	if t != nil && t.Update != nil {
+	if t.Update != nil {
 		t.Update(l, n, err)
 	}
 }
 
 type traceKey struct{}
 
-// WithTrace attaches a Trace to the context for transfer progress reporting.
+// WithTrace adds a trace to the context for transfer progress reporting.
 func WithTrace(ctx context.Context, t *Trace) context.Context {
-	return context.WithValue(ctx, traceKey{}, t)
+	old := traceFromContext(ctx)
+	if old == t {
+		// No change, return the original context. This also prevents
+		// infinite recursion below, if the caller passes the same
+		// Trace.
+		return ctx
+	}
+
+	// Create a new Trace that wraps the old one, if any. If we used the
+	// same pointer t, we end up with a recursive structure.
+	composed := &Trace{
+		Update: func(l *Layer, n int64, err error) {
+			if old != nil {
+				old.update(l, n, err)
+			}
+			t.update(l, n, err)
+		},
+	}
+	return context.WithValue(ctx, traceKey{}, composed)
 }
 
-// traceFromContext returns the Trace associated with ctx, or nil if none.
+var emptyTrace = &Trace{}
+
+// traceFromContext returns the Trace associated with ctx, or an empty Trace if
+// none is found.
+//
+// It never returns nil.
 func traceFromContext(ctx context.Context) *Trace {
 	t, _ := ctx.Value(traceKey{}).(*Trace)
+	if t == nil {
+		return emptyTrace
+	}
 	return t
 }

server/internal/internal/backoff/backoff.go

@@ -2,46 +2,44 @@ package backoff
 
 import (
 	"context"
+	"iter"
 	"math/rand/v2"
 	"time"
 )
 
-// Retry calls fn repeatedly with exponential backoff until it returns nil,
-// a non-retryable error (shouldRetry returns false), or the context is cancelled.
-// The shouldRetry function determines if an error is retryable.
-// Returns the last error encountered, or nil if fn succeeded.
-func Retry(ctx context.Context, maxBackoff time.Duration, shouldRetry func(error) bool, fn func() error) error {
-	var t *time.Timer
-	for n := 0; ; n++ {
-		if err := ctx.Err(); err != nil {
-			return err
-		}
+func Loop(ctx context.Context, maxBackoff time.Duration) iter.Seq2[int, error] {
+	var n int
+	return func(yield func(int, error) bool) {
+		var t *time.Timer
+		for {
+			if ctx.Err() != nil {
+				yield(n, ctx.Err())
+				return
+			}
 
-		err := fn()
-		if err == nil {
-			return nil
-		}
-		if !shouldRetry(err) {
-			return err
-		}
+			if !yield(n, nil) {
+				return
+			}
 
-		// n^2 backoff timer is a little smoother than the
-		// common choice of 2^n.
-		d := min(time.Duration(n*n)*10*time.Millisecond, maxBackoff)
-		// Randomize the delay between 0.5-1.5 x msec, in order
-		// to prevent accidental "thundering herd" problems.
-		d = time.Duration(float64(d) * (rand.Float64() + 0.5))
+			n++
 
-		if t == nil {
-			t = time.NewTimer(d)
-		} else {
-			t.Reset(d)
-		}
-		select {
-		case <-ctx.Done():
-			t.Stop()
-			return ctx.Err()
-		case <-t.C:
+			// n^2 backoff timer is a little smoother than the
+			// common choice of 2^n.
+			d := min(time.Duration(n*n)*10*time.Millisecond, maxBackoff)
+			// Randomize the delay between 0.5-1.5 x msec, in order
+			// to prevent accidental "thundering herd" problems.
+			d = time.Duration(float64(d) * (rand.Float64() + 0.5))
+
+			if t == nil {
+				t = time.NewTimer(d)
+			} else {
+				t.Reset(d)
+			}
+			select {
+			case <-ctx.Done():
+				t.Stop()
+			case <-t.C:
+			}
 		}
 	}
 }

server/internal/internal/backoff/backoff_synctest_test.go

@@ -10,70 +10,31 @@ import (
 	"time"
 )
 
-func TestRetry(t *testing.T) {
+func TestLoop(t *testing.T) {
 	synctest.Run(func() {
-		n := 0
+		last := -1
 
 		ctx, cancel := context.WithCancel(t.Context())
 		defer cancel()
 
-		err := Retry(ctx, 100*time.Millisecond, func(err error) bool { return true }, func() error {
-			n++
+		for n, err := range Loop(ctx, 100*time.Millisecond) {
+			if !errors.Is(err, ctx.Err()) {
+				t.Errorf("err = %v, want nil", err)
+			}
+			if err != nil {
+				break
+			}
+			if n != last+1 {
+				t.Errorf("n = %d, want %d", n, last+1)
+			}
+			last = n
 			if n > 5 {
 				cancel()
 			}
-			return errors.New("keep going")
-		})
-
-		if !errors.Is(err, context.Canceled) {
-			t.Errorf("err = %v, want context.Canceled", err)
 		}
 
-		if n != 6 {
-			t.Errorf("n = %d, want 6", n)
-		}
-	})
-}
-
-func TestRetrySuccess(t *testing.T) {
-	synctest.Run(func() {
-		n := 0
-		err := Retry(t.Context(), 100*time.Millisecond, func(err error) bool { return true }, func() error {
-			n++
-			if n >= 3 {
-				return nil // success
-			}
-			return errors.New("retry")
-		})
-
-		if err != nil {
-			t.Errorf("err = %v, want nil", err)
-		}
-		if n != 3 {
-			t.Errorf("n = %d, want 3", n)
-		}
-	})
-}
-
-func TestRetryNonRetryable(t *testing.T) {
-	synctest.Run(func() {
-		permanent := errors.New("permanent error")
-		n := 0
-		err := Retry(t.Context(), 100*time.Millisecond, func(err error) bool {
-			return !errors.Is(err, permanent)
-		}, func() error {
-			n++
-			if n >= 2 {
-				return permanent
-			}
-			return errors.New("retry")
-		})
-
-		if !errors.Is(err, permanent) {
-			t.Errorf("err = %v, want permanent", err)
-		}
-		if n != 2 {
-			t.Errorf("n = %d, want 2", n)
+		if last != 6 {
+			t.Errorf("last = %d, want 6", last)
 		}
 	})
 }

server/internal/internal/backoff/backoff_test.go

@@ -3,46 +3,37 @@
 package backoff
 
 import (
-	"errors"
 	"testing"
 	"testing/synctest"
 	"time"
 )
 
-var errRetry = errors.New("retry")
-
-func TestRetryAllocs(t *testing.T) {
+func TestLoopAllocs(t *testing.T) {
 	for i := range 3 {
 		got := testing.AllocsPerRun(1000, func() {
-			tick := 0
-			Retry(t.Context(), 1, func(err error) bool { return true }, func() error {
-				tick++
+			for tick := range Loop(t.Context(), 1) {
 				if tick >= i {
-					return nil
+					break
 				}
-				return errRetry
-			})
+			}
 		})
 		want := float64(0)
 		if i > 0 {
 			want = 3 // due to time.NewTimer
 		}
 		if got > want {
-			t.Errorf("[%d ticks]: allocs = %v, want <= %v", i, got, want)
+			t.Errorf("[%d ticks]: allocs = %v, want 0", i, want)
 		}
 	}
 }
 
-func BenchmarkRetry(b *testing.B) {
+func BenchmarkLoop(b *testing.B) {
 	ctx := b.Context()
 	synctest.Run(func() {
-		n := 0
-		Retry(ctx, 100*time.Millisecond, func(err error) bool { return true }, func() error {
-			n++
+		for n := range Loop(ctx, 100*time.Millisecond) {
 			if n == b.N {
-				return nil
+				break
 			}
-			return errRetry
-		})
+		}
 	})
 }

server/internal/registry/server.go

@@ -231,7 +231,7 @@ func (s *Local) handleDelete(_ http.ResponseWriter, r *http.Request) error {
 	if r.Method != "DELETE" {
 		return errMethodNotAllowed
 	}
-	p, err := decodeParams(r.Body)
+	p, err := decodeUserJSON[*params](r.Body)
 	if err != nil {
 		return err
 	}
@@ -261,7 +261,7 @@ func (s *Local) handlePull(w http.ResponseWriter, r *http.Request) error {
 		return errMethodNotAllowed
 	}
 
-	p, err := decodeParams(r.Body)
+	p, err := decodeUserJSON[*params](r.Body)
 	if err != nil {
 		return err
 	}
@@ -293,14 +293,10 @@ func (s *Local) handlePull(w http.ResponseWriter, r *http.Request) error {
 		}
 	}
 
-	// ticker controls periodic progress flushing. It starts paused (very long
-	// interval) and is activated by start() once all layers are registered,
-	// so clients see a complete total before progress begins.
-	ticker := time.NewTicker(1 << 62) // effectively paused until started
-	defer ticker.Stop()
+	t := time.NewTicker(1<<63 - 1) // "unstarted" timer
 	start := sync.OnceFunc(func() {
-		flushProgress()
-		ticker.Reset(100 * time.Millisecond)
+		flushProgress() // flush initial state
+		t.Reset(100 * time.Millisecond)
 	})
 	ctx := ollama.WithTrace(r.Context(), &ollama.Trace{
 		Update: func(l *ollama.Layer, n int64, err error) {
@@ -324,21 +320,36 @@ func (s *Local) handlePull(w http.ResponseWriter, r *http.Request) error {
 				})
 			}()
 
+			// Block flushing progress updates until every
+			// layer is accounted for. Clients depend on a
+			// complete model size to calculate progress
+			// correctly; if they use an incomplete total,
+			// progress indicators would erratically jump
+			// as new layers are registered.
 			start()
 		},
 	})
 
 	done := make(chan error, 1)
-	go func() {
-		done <- backoff.Retry(ctx, 3*time.Second, canRetry, func() error {
-			return s.Client.Pull(ctx, p.model())
-		})
+	go func() (err error) {
+		defer func() { done <- err }()
+		for _, err := range backoff.Loop(ctx, 3*time.Second) {
+			if err != nil {
+				return err
+			}
+			err := s.Client.Pull(ctx, p.model())
+			if canRetry(err) {
+				continue
+			}
+			return err
+		}
+		return nil
 	}()
 
 	enc.Encode(progressUpdateJSON{Status: "pulling manifest"})
 	for {
 		select {
-		case <-ticker.C:
+		case <-t.C:
 			flushProgress()
 		case err := <-done:
 			flushProgress()
@@ -363,13 +374,20 @@ func (s *Local) handlePull(w http.ResponseWriter, r *http.Request) error {
 	}
 }
 
-func decodeParams(r io.Reader) (*params, error) {
-	var p params
-	err := json.NewDecoder(r).Decode(&p)
+func decodeUserJSON[T any](r io.Reader) (T, error) {
+	var v T
+	err := json.NewDecoder(r).Decode(&v)
 	if err == nil {
-		return &p, nil
+		return v, nil
 	}
+	var zero T
 
+	// Not sure why, but I can't seem to be able to use:
+	//
+	//   errors.As(err, &json.UnmarshalTypeError{})
+	//
+	// This is working fine in stdlib, so I'm not sure what rules changed
+	// and why this no longer works here. So, we do it the verbose way.
 	var a *json.UnmarshalTypeError
 	var b *json.SyntaxError
 	if errors.As(err, &a) || errors.As(err, &b) {
@@ -378,7 +396,7 @@ func decodeParams(r io.Reader) (*params, error) {
 	if errors.Is(err, io.EOF) {
 		err = &serverError{Status: 400, Message: "empty request body", Code: "bad_request"}
 	}
-	return nil, err
+	return zero, err
 }
 
 func canRetry(err error) bool {
@@ -390,8 +408,10 @@ func canRetry(err error) bool {
 		return oe.Temporary()
 	}
 	s := err.Error()
-	return errors.Is(err, context.DeadlineExceeded) ||
-		strings.Contains(s, "unreachable") ||
-		strings.Contains(s, "no route to host") ||
-		strings.Contains(s, "connection reset by peer")
+	return cmp.Or(
+		errors.Is(err, context.DeadlineExceeded),
+		strings.Contains(s, "unreachable"),
+		strings.Contains(s, "no route to host"),
+		strings.Contains(s, "connection reset by peer"),
+	)
 }

server/sparse_common.go

@@ -0,0 +1,8 @@
+//go:build !windows
+
+package server
+
+import "os"
+
+func setSparse(*os.File) {
+}

server/sparse_windows.go

@@ -0,0 +1,17 @@
+package server
+
+import (
+	"os"
+
+	"golang.org/x/sys/windows"
+)
+
+func setSparse(file *os.File) {
+	// exFat (and other FS types) don't support sparse files, so ignore errors
+	windows.DeviceIoControl( //nolint:errcheck
+		windows.Handle(file.Fd()), windows.FSCTL_SET_SPARSE,
+		nil, 0,
+		nil, 0,
+		nil, nil,
+	)
+}