mirror of
https://github.com/ollama/ollama.git
synced 2026-10-14 11:24:10 +00:00
transfer: return disk-full errors during blob download (fixes ollama/ollama#18644)
The MLX pull path ignored errors from writing a blob, so a full disk still counted the blob as downloaded. "no space left" only appeared later, while writing the manifest, after the payload had been read. Return the write error immediately and do not retry it. Co-authored-by: Ganesh <ganeshmaheshwaram2005@gmail.com>
This commit is contained in:
parent
8a971df3ba
commit
b933ca379d
3 changed files with 118 additions and 5 deletions
|
|
@ -26,6 +26,18 @@ var (
|
|||
errSlow = errors.New("download too slow")
|
||||
)
|
||||
|
||||
// destError is a write to the blob destination. It is not a transient
|
||||
// network failure, so the download is not retried.
|
||||
type destError struct{ err error }
|
||||
|
||||
func (e *destError) Error() string { return e.err.Error() }
|
||||
func (e *destError) Unwrap() error { return e.err }
|
||||
|
||||
func isDestWriteError(err error) bool {
|
||||
var de *destError
|
||||
return errors.As(err, &de)
|
||||
}
|
||||
|
||||
type downloader struct {
|
||||
client *http.Client
|
||||
baseURL string
|
||||
|
|
@ -206,6 +218,10 @@ func (d *downloader) download(ctx context.Context, blob Blob) error {
|
|||
switch {
|
||||
case errors.Is(err, context.Canceled), errors.Is(err, context.DeadlineExceeded):
|
||||
return err
|
||||
case isDestWriteError(err):
|
||||
// Disk full and other local write failures will not succeed on retry,
|
||||
// and a later network error would hide them.
|
||||
return err
|
||||
case errors.Is(err, errStalled):
|
||||
if stallRetries++; stallRetries >= maxTransientRetries {
|
||||
attempt++
|
||||
|
|
@ -336,7 +352,10 @@ func (d *downloader) save(ctx context.Context, cancel context.CancelCauseFunc, b
|
|||
// Don't remove .tmp here — download() handles cleanup based on blob size
|
||||
return existingSize + n, err
|
||||
}
|
||||
f.Close()
|
||||
// The final flush can be where a full disk reports the error.
|
||||
if err := f.Close(); err != nil {
|
||||
return existingSize + n, &destError{err: err}
|
||||
}
|
||||
|
||||
if got := fmt.Sprintf("sha256:%x", h.Sum(nil)); got != blob.Digest {
|
||||
os.Remove(tmp)
|
||||
|
|
@ -399,10 +418,21 @@ func (d *downloader) copy(ctx context.Context, cancel context.CancelCauseFunc, d
|
|||
nr, err := src.Read(buf)
|
||||
if nr > 0 {
|
||||
lastRead.Store(time.Now().UnixNano())
|
||||
dst.Write(buf[:nr])
|
||||
h.Write(buf[:nr])
|
||||
d.progress.add(int64(nr))
|
||||
n.Add(int64(nr))
|
||||
// A full disk fails here. Dropping the error reports the blob as
|
||||
// downloaded and only surfaces "no space left" later, while writing
|
||||
// the manifest, after the payload has already been read.
|
||||
nw, werr := dst.Write(buf[:nr])
|
||||
if nw > 0 {
|
||||
h.Write(buf[:nw])
|
||||
d.progress.add(int64(nw))
|
||||
n.Add(int64(nw))
|
||||
}
|
||||
if werr != nil {
|
||||
return n.Load(), &destError{err: werr}
|
||||
}
|
||||
if nw < nr {
|
||||
return n.Load(), &destError{err: io.ErrShortWrite}
|
||||
}
|
||||
}
|
||||
if err == io.EOF {
|
||||
return n.Load(), nil
|
||||
|
|
|
|||
62
transfer/download_nospace_unix_test.go
Normal file
62
transfer/download_nospace_unix_test.go
Normal file
|
|
@ -0,0 +1,62 @@
|
|||
//go:build unix
|
||||
|
||||
package transfer
|
||||
|
||||
import (
|
||||
"context"
|
||||
"crypto/sha256"
|
||||
"errors"
|
||||
"fmt"
|
||||
"net/http"
|
||||
"net/http/httptest"
|
||||
"os"
|
||||
"path/filepath"
|
||||
"sync/atomic"
|
||||
"syscall"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// Regression for ollama/ollama#18644: a full disk must fail the blob write
|
||||
// itself. Previously the write error was discarded, the blob was treated as
|
||||
// downloaded, and "no space left" only appeared later while writing the
|
||||
// manifest — after the whole payload had been read.
|
||||
func TestDownloadNoSpace(t *testing.T) {
|
||||
if _, err := os.Stat("/dev/full"); err != nil {
|
||||
t.Skip("/dev/full not available")
|
||||
}
|
||||
|
||||
payload := []byte("blob-bytes")
|
||||
sum := sha256.Sum256(payload)
|
||||
digest := fmt.Sprintf("sha256:%x", sum)
|
||||
|
||||
var hits atomic.Int32
|
||||
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
|
||||
hits.Add(1)
|
||||
w.WriteHeader(http.StatusOK)
|
||||
_, _ = w.Write(payload)
|
||||
}))
|
||||
t.Cleanup(server.Close)
|
||||
|
||||
clientDir := t.TempDir()
|
||||
tmp := filepath.Join(clientDir, digestToPath(digest)+".tmp")
|
||||
if err := os.MkdirAll(filepath.Dir(tmp), 0o755); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
// /dev/full accepts the open and returns ENOSPC on write.
|
||||
if err := os.Symlink("/dev/full", tmp); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
|
||||
err := Download(context.Background(), DownloadOptions{
|
||||
Blobs: []Blob{{Digest: digest, Size: int64(len(payload))}},
|
||||
BaseURL: server.URL,
|
||||
DestDir: clientDir,
|
||||
})
|
||||
if !errors.Is(err, syscall.ENOSPC) {
|
||||
t.Fatalf("Download error = %v, want ENOSPC", err)
|
||||
}
|
||||
// resolve + one body GET. A retried disk-full error would hit the server again.
|
||||
if got := hits.Load(); got != 2 {
|
||||
t.Fatalf("server hits = %d, want 2 (one attempt)", got)
|
||||
}
|
||||
}
|
||||
|
|
@ -457,6 +457,27 @@ func TestDownloadDigestMismatch(t *testing.T) {
|
|||
}
|
||||
}
|
||||
|
||||
// A destination that cannot accept bytes must fail the copy. Ignoring the
|
||||
// write error lets the downloader treat the blob as complete.
|
||||
func TestCopyReturnsDestinationWriteError(t *testing.T) {
|
||||
errFull := errors.New("no space left on device")
|
||||
d := &downloader{
|
||||
progress: newProgressTracker(4, func(int64, int64) {}),
|
||||
stallTimeout: time.Hour,
|
||||
}
|
||||
n, err := d.copy(context.Background(), func(error) {}, errWriter{err: errFull}, strings.NewReader("blob"), sha256.New())
|
||||
if !errors.Is(err, errFull) {
|
||||
t.Fatalf("copy error = %v, want no space left", err)
|
||||
}
|
||||
if n != 0 {
|
||||
t.Fatalf("copy counted %d bytes after a failed write", n)
|
||||
}
|
||||
}
|
||||
|
||||
type errWriter struct{ err error }
|
||||
|
||||
func (w errWriter) Write([]byte) (int, error) { return 0, w.err }
|
||||
|
||||
func TestUpload(t *testing.T) {
|
||||
// Create test blobs
|
||||
clientDir := t.TempDir()
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue