soft_quay/core/downloader/queue_test.go

package downloader_test

import (
	"bytes"
	"context"
	"errors"
	"io"
	"net/http"
	"net/http/httptest"
	"os"
	"sync"
	"testing"
	"time"

	"softbox.local/core/downloader"
	"softbox.local/core/storage"
)

func TestQueueDefaultConcurrencyDoesNotExceedTwo(t *testing.T) {
	root := t.TempDir()
	transport := newBlockingTransport([]byte("data"))
	observer := newEventObserver()
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     transport,
		Observer:      observer,
		ProgressEvery: time.Hour,
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())

	urls := make([]string, 5)
	for index := range urls {
		urls[index] = "https://download.invalid/request-" + string(rune('a'+index))
		_, err := queue.Enqueue(context.Background(), downloader.EnqueueRequest{
			RequestID:  "request-" + string(rune('a'+index)),
			AppID:      "app-" + string(rune('a'+index)),
			URL:        urls[index],
			TotalKnown: true,
			Total:      4,
		})
		if err != nil {
			t.Fatalf("Enqueue(%d) error = %v", index, err)
		}
	}

	first := waitOpen(t, transport.opens)
	second := waitOpen(t, transport.opens)
	select {
	case third := <-transport.opens:
		t.Fatalf("third transfer started early: %#v", third)
	case <-time.After(100 * time.Millisecond):
	}
	if transport.maxActiveValue() != 2 {
		t.Fatalf("max active = %d, want 2", transport.maxActiveValue())
	}

	transport.release(first.URL)
	_ = waitOpen(t, transport.opens)
	transport.release(second.URL)
}

func TestQueueActivePauseStartsNextQueuedTask(t *testing.T) {
	root := t.TempDir()
	transport := newBlockingTransport([]byte("data"))
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     transport,
		Observer:      newEventObserver(),
		MaxConcurrent: 2,
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	for _, suffix := range []string{"a", "b", "c"} {
		_, err := queue.Enqueue(context.Background(), downloader.EnqueueRequest{
			RequestID:  "active-pause-" + suffix,
			AppID:      "active-pause-app-" + suffix,
			URL:        "https://download.invalid/active-pause-" + suffix,
			TotalKnown: true,
			Total:      4,
		})
		if err != nil {
			t.Fatalf("Enqueue(%s) error = %v", suffix, err)
		}
	}
	first := waitOpen(t, transport.opens)
	second := waitOpen(t, transport.opens)
	initialURLs := map[string]bool{
		first.URL:  true,
		second.URL: true,
	}
	if !initialURLs["https://download.invalid/active-pause-a"] ||
		!initialURLs["https://download.invalid/active-pause-b"] ||
		len(initialURLs) != 2 {
		t.Fatalf("initial opens = %#v, %#v", first, second)
	}
	if err := queue.Pause(context.Background(), "active-pause-a"); err != nil {
		t.Fatalf("Pause(active) error = %v", err)
	}
	next := waitOpen(t, transport.opens)
	if next.URL != "https://download.invalid/active-pause-c" {
		t.Fatalf("next URL = %q, want queued task", next.URL)
	}
	paused, found := queue.Snapshot("active-pause-a")
	if !found || paused.Status != downloader.StatusPaused {
		t.Fatalf("paused task = %#v, found=%t", paused, found)
	}
	if transport.maxActiveValue() > 2 {
		t.Fatalf("max active = %d, want <= 2", transport.maxActiveValue())
	}
}

func TestQueuePauseReleasesSlotAndQueuedPauseDoesNotStart(t *testing.T) {
	root := t.TempDir()
	transport := newBlockingTransport([]byte("data"))
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     transport,
		Observer:      newEventObserver(),
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	for _, suffix := range []string{"a", "b", "c"} {
		_, err := queue.Enqueue(context.Background(), downloader.EnqueueRequest{
			RequestID:  "pause-" + suffix,
			AppID:      "pause-app-" + suffix,
			URL:        "https://download.invalid/pause-" + suffix,
			TotalKnown: true,
			Total:      4,
		})
		if err != nil {
			t.Fatalf("Enqueue(%s) error = %v", suffix, err)
		}
	}
	_ = waitOpen(t, transport.opens)
	_ = waitOpen(t, transport.opens)
	if err := queue.Pause(context.Background(), "pause-c"); err != nil {
		t.Fatalf("Pause(queued) error = %v", err)
	}
	if err := queue.Pause(context.Background(), "pause-a"); err != nil {
		t.Fatalf("Pause(active) error = %v", err)
	}
	select {
	case request := <-transport.opens:
		t.Fatalf("paused queued task started: %#v", request)
	case <-time.After(100 * time.Millisecond):
	}
	if err := queue.Resume(context.Background(), "pause-c"); err != nil {
		t.Fatalf("Resume() error = %v", err)
	}
	started := waitOpen(t, transport.opens)
	if started.URL != "https://download.invalid/pause-c" {
		t.Fatalf("started URL = %q", started.URL)
	}
}

func TestQueuePauseResumeUsesDiskOffsetAndValidator(t *testing.T) {
	root := t.TempDir()
	transport := &resumeTransport{}
	observer := newEventObserver()
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     transport,
		Observer:      observer,
		Clock:         newAdvancingClock(),
		ProgressEvery: downloader.DefaultProgressPeriod,
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())

	_, err = queue.Enqueue(context.Background(), downloader.EnqueueRequest{
		RequestID:  "request-resume",
		AppID:      "app-resume",
		URL:        "https://download.invalid/resume",
		TotalKnown: true,
		Total:      4,
	})
	if err != nil {
		t.Fatalf("Enqueue() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventProgress, "request-resume")
	if err := queue.Pause(context.Background(), "request-resume"); err != nil {
		t.Fatalf("Pause() error = %v", err)
	}
	task, found := queue.Snapshot("request-resume")
	if !found || task.Status != downloader.StatusPaused || task.Done != 2 {
		t.Fatalf("paused task = %#v, found=%t", task, found)
	}
	assertNoEvent(t, observer.events, downloader.EventProgress, "request-resume")
	if err := queue.Resume(context.Background(), "request-resume"); err != nil {
		t.Fatalf("Resume() error = %v", err)
	}
	completed := waitEvent(
		t,
		observer.events,
		downloader.EventCompleted,
		"request-resume",
	)
	if completed.Done != 4 {
		t.Fatalf("completed done = %d, want 4", completed.Done)
	}
	opens := transport.openRequests()
	if len(opens) != 2 ||
		opens[0].Offset != 0 ||
		opens[1].Offset != 2 ||
		opens[1].Validator.ETag != `"v1"` {
		t.Fatalf("open requests = %#v", opens)
	}
}

func TestQueuePauseReportsBodyCloseFailure(t *testing.T) {
	root := t.TempDir()
	transport := &closeErrorTransport{
		opened: make(chan struct{}, 1),
	}
	observer := newEventObserver()
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     transport,
		Observer:      observer,
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	request := downloader.EnqueueRequest{
		RequestID:  "request-close-error",
		AppID:      "app-close-error",
		URL:        "https://download.invalid/close-error",
		TotalKnown: true,
		Total:      4,
	}
	if _, err := queue.Enqueue(context.Background(), request); err != nil {
		t.Fatalf("Enqueue() error = %v", err)
	}
	select {
	case <-transport.opened:
	case <-time.After(3 * time.Second):
		t.Fatal("timed out waiting for close-error transfer")
	}
	if err := queue.Pause(context.Background(), request.RequestID); !errors.Is(
		err,
		errInjectedBodyClose,
	) {
		t.Fatalf("Pause() error = %v, want %v", err, errInjectedBodyClose)
	}
	failed, found := queue.Snapshot(request.RequestID)
	if !found || failed.Status != downloader.StatusFailed {
		t.Fatalf("failed task = %#v, found=%t", failed, found)
	}
	waitEvent(t, observer.events, downloader.EventFailed, request.RequestID)
}

func TestQueueResumeFinalizesCompletePausedPartWithoutReopening(t *testing.T) {
	root := t.TempDir()
	transport := newFullThenBlockTransport([]byte("data"))
	observer := newEventObserver()
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     transport,
		Observer:      observer,
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	request := downloader.EnqueueRequest{
		RequestID:  "request-full-pause",
		AppID:      "app-full-pause",
		URL:        "https://download.invalid/full-pause",
		TotalKnown: true,
		Total:      4,
	}
	if _, err := queue.Enqueue(context.Background(), request); err != nil {
		t.Fatalf("Enqueue() error = %v", err)
	}
	transport.waitWritten(t)
	if err := queue.Pause(context.Background(), request.RequestID); err != nil {
		t.Fatalf("Pause() error = %v", err)
	}
	paused, found := queue.Snapshot(request.RequestID)
	if !found || paused.Status != downloader.StatusPaused || paused.Done != 4 {
		t.Fatalf("paused task = %#v, found=%t", paused, found)
	}
	if err := queue.Resume(context.Background(), request.RequestID); err != nil {
		t.Fatalf("Resume() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventCompleted, request.RequestID)
	if transport.callsValue() != 1 {
		t.Fatalf("transport calls = %d, want 1", transport.callsValue())
	}
}

func TestQueueRetryResumesAfterInterruptedBody(t *testing.T) {
	root := t.TempDir()
	transport := &interruptedTransport{}
	observer := newEventObserver()
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     transport,
		Observer:      observer,
		Clock:         newAdvancingClock(),
		ProgressEvery: downloader.DefaultProgressPeriod,
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	request := downloader.EnqueueRequest{
		RequestID:  "request-interrupted",
		AppID:      "app-interrupted",
		URL:        "https://download.invalid/interrupted",
		TotalKnown: true,
		Total:      4,
	}
	if _, err := queue.Enqueue(context.Background(), request); err != nil {
		t.Fatalf("Enqueue() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventFailed, request.RequestID)
	failed, _ := queue.Snapshot(request.RequestID)
	if failed.Done != 2 {
		t.Fatalf("failed Done = %d, want 2", failed.Done)
	}
	if err := queue.Retry(context.Background(), request.RequestID); err != nil {
		t.Fatalf("Retry() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventCompleted, request.RequestID)
	opens := transport.openRequests()
	if len(opens) != 2 || opens[1].Offset != 2 ||
		opens[1].Validator.ETag != `"v1"` {
		t.Fatalf("open requests = %#v", opens)
	}
}

func TestQueueRetryFinalizesCompleteFailedPartWithoutReopening(t *testing.T) {
	root := t.TempDir()
	transport := &fullThenErrorTransport{data: []byte("data")}
	observer := newEventObserver()
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     transport,
		Observer:      observer,
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	request := downloader.EnqueueRequest{
		RequestID:  "request-full-failure",
		AppID:      "app-full-failure",
		URL:        "https://download.invalid/full-failure",
		TotalKnown: true,
		Total:      4,
	}
	if _, err := queue.Enqueue(context.Background(), request); err != nil {
		t.Fatalf("Enqueue() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventFailed, request.RequestID)
	failed, found := queue.Snapshot(request.RequestID)
	if !found || failed.Status != downloader.StatusFailed || failed.Done != 4 {
		t.Fatalf("failed task = %#v, found=%t", failed, found)
	}
	if err := queue.Retry(context.Background(), request.RequestID); err != nil {
		t.Fatalf("Retry() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventCompleted, request.RequestID)
	if transport.callsValue() != 1 {
		t.Fatalf("transport calls = %d, want 1", transport.callsValue())
	}
}

func TestQueueHTTPConnectionInterruptionResumesWithRange(t *testing.T) {
	content := []byte("data")
	var mu sync.Mutex
	var ranges []string
	server := httptest.NewTLSServer(http.HandlerFunc(func(
		writer http.ResponseWriter,
		request *http.Request,
	) {
		mu.Lock()
		ranges = append(ranges, request.Header.Get("Range"))
		call := len(ranges)
		mu.Unlock()
		writer.Header().Set("ETag", `"v1"`)
		if call == 1 {
			writer.Header().Set("Content-Length", "4")
			_, _ = writer.Write(content[:2])
			return
		}
		if request.Header.Get("Range") != "bytes=2-" ||
			request.Header.Get("If-Range") != `"v1"` {
			t.Errorf(
				"resume headers Range=%q If-Range=%q",
				request.Header.Get("Range"),
				request.Header.Get("If-Range"),
			)
		}
		writer.Header().Set("Content-Range", "bytes 2-3/4")
		writer.Header().Set("Content-Length", "2")
		writer.WriteHeader(http.StatusPartialContent)
		_, _ = writer.Write(content[2:])
	}))
	defer server.Close()

	root := t.TempDir()
	observer := newEventObserver()
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     downloader.NewHTTPTransport(server.Client()),
		Observer:      observer,
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	request := downloader.EnqueueRequest{
		RequestID:  "request-http-resume",
		AppID:      "app-http-resume",
		URL:        server.URL,
		TotalKnown: true,
		Total:      4,
	}
	if _, err := queue.Enqueue(context.Background(), request); err != nil {
		t.Fatalf("Enqueue() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventFailed, request.RequestID)
	if err := queue.Retry(context.Background(), request.RequestID); err != nil {
		t.Fatalf("Retry() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventCompleted, request.RequestID)
	paths, _ := downloader.DeriveTaskPaths(root, request.RequestID)
	document, err := os.ReadFile(paths.Completed)
	if err != nil {
		t.Fatalf("ReadFile(completed) error = %v", err)
	}
	if !bytes.Equal(document, content) {
		t.Fatalf("completed bytes = %q", document)
	}
}

func TestQueueRangeIgnoredStartsNewAttemptAndReplacesPart(t *testing.T) {
	root := t.TempDir()
	store := storage.NewDownloadTaskStore(root)
	task := downloader.Task{
		SchemaVersion: downloader.TaskSchemaVersion,
		RequestID:     "request-restart",
		AppID:         "app-restart",
		URL:           "https://download.invalid/restart",
		Status:        downloader.StatusPaused,
		Attempt:       1,
		Done:          2,
		TotalKnown:    true,
		Total:         4,
		Validator:     downloader.EntityValidator{ETag: `"v1"`},
		CreatedAt:     "2026-07-16T00:00:00Z",
	}
	if err := store.Write(task); err != nil {
		t.Fatalf("Write() error = %v", err)
	}
	paths, _ := downloader.DeriveTaskPaths(root, task.RequestID)
	if err := os.WriteFile(paths.Part, []byte("ab"), 0o600); err != nil {
		t.Fatalf("WriteFile(part) error = %v", err)
	}
	observer := newEventObserver()
	transport := &restartTransport{}
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         store,
		Transport:     transport,
		Observer:      observer,
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	if err := queue.Resume(context.Background(), task.RequestID); err != nil {
		t.Fatalf("Resume() error = %v", err)
	}
	completed := waitEvent(
		t,
		observer.events,
		downloader.EventCompleted,
		task.RequestID,
	)
	if completed.Attempt != 3 {
		t.Fatalf("completed attempt = %d, want 3", completed.Attempt)
	}
	var startedAttempts []uint64
	for _, event := range observer.snapshot() {
		if event.RequestID == task.RequestID && event.Type == downloader.EventStarted {
			startedAttempts = append(startedAttempts, event.Attempt)
		}
	}
	if len(startedAttempts) != 2 ||
		startedAttempts[0] != 2 ||
		startedAttempts[1] != 3 {
		t.Fatalf("started attempts = %#v, want [2 3]", startedAttempts)
	}
	document, err := os.ReadFile(paths.Completed)
	if err != nil {
		t.Fatalf("ReadFile(completed) error = %v", err)
	}
	if string(document) != "WXYZ" {
		t.Fatalf("completed bytes = %q, want WXYZ", document)
	}
}

func TestQueueRestartTruncatesPartBeforePersistingNewAttempt(t *testing.T) {
	root := t.TempDir()
	inner := storage.NewDownloadTaskStore(root)
	task := downloader.Task{
		SchemaVersion: downloader.TaskSchemaVersion,
		RequestID:     "request-restart-write-failure",
		AppID:         "app-restart-write-failure",
		URL:           "https://download.invalid/restart-write-failure",
		Status:        downloader.StatusPaused,
		Attempt:       1,
		Done:          2,
		TotalKnown:    true,
		Total:         4,
		Validator:     downloader.EntityValidator{ETag: `"v1"`},
		CreatedAt:     "2026-07-16T00:00:00Z",
	}
	if err := inner.Write(task); err != nil {
		t.Fatalf("Write() error = %v", err)
	}
	paths, _ := downloader.DeriveTaskPaths(root, task.RequestID)
	if err := os.WriteFile(paths.Part, []byte("ab"), 0o600); err != nil {
		t.Fatalf("WriteFile(part) error = %v", err)
	}
	observer := newEventObserver()
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store: &restartWriteFailStore{
			inner: inner,
		},
		Transport: &restartTransport{},
		Observer:  observer,
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	if err := queue.Resume(context.Background(), task.RequestID); err != nil {
		t.Fatalf("Resume() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventFailed, task.RequestID)
	partInfo, err := os.Stat(paths.Part)
	if err != nil {
		t.Fatalf("Stat(part) error = %v", err)
	}
	if partInfo.Size() != 0 {
		t.Fatalf("part size = %d, want 0 before restart metadata write", partInfo.Size())
	}
	if err := queue.Close(context.Background()); err != nil {
		t.Fatalf("Close(first queue) error = %v", err)
	}

	retryObserver := newEventObserver()
	retryTransport := &recordingFullTransport{data: []byte("WXYZ")}
	recovered, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         inner,
		Transport:     retryTransport,
		Observer:      retryObserver,
	})
	if err != nil {
		t.Fatalf("NewQueue(recovered) error = %v", err)
	}
	defer recovered.Close(context.Background())
	if err := recovered.Retry(context.Background(), task.RequestID); err != nil {
		t.Fatalf("Retry() error = %v", err)
	}
	waitEvent(t, retryObserver.events, downloader.EventCompleted, task.RequestID)
	opens := retryTransport.openRequests()
	if len(opens) != 1 || opens[0].Offset != 0 {
		t.Fatalf("recovered opens = %#v, want one offset=0 request", opens)
	}
}

func TestQueueCancelCleansActiveTaskAndSuppressesLateProgress(t *testing.T) {
	root := t.TempDir()
	transport := newBlockingTransport([]byte("data"))
	observer := newEventObserver()
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     transport,
		Observer:      observer,
		ProgressEvery: 0,
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	request := downloader.EnqueueRequest{
		RequestID:  "request-cancel",
		AppID:      "app-cancel",
		URL:        "https://download.invalid/cancel",
		TotalKnown: true,
		Total:      4,
	}
	if _, err := queue.Enqueue(context.Background(), request); err != nil {
		t.Fatalf("Enqueue() error = %v", err)
	}
	_ = waitOpen(t, transport.opens)
	if err := queue.Cancel(context.Background(), request.RequestID); err != nil {
		t.Fatalf("Cancel() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventCanceled, request.RequestID)
	if _, found := queue.Snapshot(request.RequestID); found {
		t.Fatal("canceled task still exists")
	}
	paths, _ := downloader.DeriveTaskPaths(root, request.RequestID)
	for _, path := range []string{paths.Part, paths.Completed, paths.Metadata} {
		if _, err := os.Stat(path); !os.IsNotExist(err) {
			t.Fatalf("canceled path still exists %s: %v", path, err)
		}
	}
	assertNoEvent(t, observer.events, downloader.EventProgress, request.RequestID)
}

func TestQueueCancelRemainsSuccessfulWhenObserverFails(t *testing.T) {
	root := t.TempDir()
	transport := newBlockingTransport([]byte("data"))
	observer := newFailingEventObserver(downloader.EventCanceled)
	reported := make(chan downloader.Event, 1)
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     transport,
		Observer:      observer,
		OnObserverError: func(event downloader.Event, _ error) {
			reported <- event
		},
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	request := downloader.EnqueueRequest{
		RequestID:  "request-cancel-observer-failure",
		AppID:      "app-cancel-observer-failure",
		URL:        "https://download.invalid/cancel-observer-failure",
		TotalKnown: true,
		Total:      4,
	}
	if _, err := queue.Enqueue(context.Background(), request); err != nil {
		t.Fatalf("Enqueue() error = %v", err)
	}
	_ = waitOpen(t, transport.opens)
	if err := queue.Cancel(context.Background(), request.RequestID); err != nil {
		t.Fatalf("Cancel() error = %v", err)
	}
	if _, found := queue.Snapshot(request.RequestID); found {
		t.Fatal("canceled task still exists after observer failure")
	}
	select {
	case event := <-reported:
		if event.Type != downloader.EventCanceled ||
			event.RequestID != request.RequestID {
			t.Fatalf("reported observer event = %#v", event)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("observer failure was not reported")
	}
	paths, _ := downloader.DeriveTaskPaths(root, request.RequestID)
	for _, path := range []string{paths.Part, paths.Completed, paths.Metadata} {
		if _, err := os.Stat(path); !os.IsNotExist(err) {
			t.Fatalf("canceled path still exists %s: %v", path, err)
		}
	}
	replacement := request
	replacement.RequestID = "request-cancel-observer-replacement"
	replacement.URL = "https://download.invalid/cancel-observer-replacement"
	if _, err := queue.Enqueue(context.Background(), replacement); err != nil {
		t.Fatalf("Enqueue(replacement) error = %v", err)
	}
}

func TestQueueCancelCleansTaskDespiteBodyCloseFailure(t *testing.T) {
	root := t.TempDir()
	transport := &closeErrorTransport{
		opened: make(chan struct{}, 1),
	}
	observer := newEventObserver()
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     transport,
		Observer:      observer,
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	request := downloader.EnqueueRequest{
		RequestID:  "request-cancel-close-error",
		AppID:      "app-cancel-close-error",
		URL:        "https://download.invalid/cancel-close-error",
		TotalKnown: true,
		Total:      4,
	}
	if _, err := queue.Enqueue(context.Background(), request); err != nil {
		t.Fatalf("Enqueue() error = %v", err)
	}
	select {
	case <-transport.opened:
	case <-time.After(3 * time.Second):
		t.Fatal("timed out waiting for cancel close-error transfer")
	}
	if err := queue.Cancel(context.Background(), request.RequestID); err != nil {
		t.Fatalf("Cancel() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventCanceled, request.RequestID)
	if _, found := queue.Snapshot(request.RequestID); found {
		t.Fatal("canceled task still exists")
	}
	paths, _ := downloader.DeriveTaskPaths(root, request.RequestID)
	for _, path := range []string{paths.Part, paths.Completed, paths.Metadata} {
		if _, err := os.Stat(path); !os.IsNotExist(err) {
			t.Fatalf("canceled path still exists %s: %v", path, err)
		}
	}
	assertNoEvent(t, observer.events, downloader.EventFailed, request.RequestID)
}

func TestQueueProgressObserverFailureDoesNotFailTransfer(t *testing.T) {
	root := t.TempDir()
	observer := newFailingEventObserver(downloader.EventProgress)
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     &recordingFullTransport{data: []byte("data")},
		Observer:      observer,
		ProgressEvery: 0,
		OnObserverError: func(downloader.Event, error) {
		},
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	request := downloader.EnqueueRequest{
		RequestID:  "request-progress-observer-failure",
		AppID:      "app-progress-observer-failure",
		URL:        "https://download.invalid/progress-observer-failure",
		TotalKnown: true,
		Total:      4,
	}
	if _, err := queue.Enqueue(context.Background(), request); err != nil {
		t.Fatalf("Enqueue() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventCompleted, request.RequestID)
	completed, found := queue.Snapshot(request.RequestID)
	if !found || completed.Status != downloader.StatusCompleted {
		t.Fatalf("completed task = %#v, found=%t", completed, found)
	}
}

func TestQueueCompletedObserverFailureIsReportedAndRecoverable(t *testing.T) {
	root := t.TempDir()
	observer := newFailingEventObserver(downloader.EventCompleted)
	reported := make(chan downloader.Event, 1)
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     &recordingFullTransport{data: []byte("data")},
		Observer:      observer,
		OnObserverError: func(event downloader.Event, _ error) {
			reported <- event
		},
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	request := downloader.EnqueueRequest{
		RequestID:  "request-completed-observer-failure",
		AppID:      "app-completed-observer-failure",
		URL:        "https://download.invalid/completed-observer-failure",
		TotalKnown: true,
		Total:      4,
	}
	if _, err := queue.Enqueue(context.Background(), request); err != nil {
		t.Fatalf("Enqueue() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventCompleted, request.RequestID)
	select {
	case event := <-reported:
		if event.Type != downloader.EventCompleted ||
			event.RequestID != request.RequestID {
			t.Fatalf("reported observer event = %#v", event)
		}
	case <-time.After(3 * time.Second):
		t.Fatal("completed observer failure was not reported")
	}
	tasks := queue.Tasks()
	if len(tasks) != 1 ||
		tasks[0].RequestID != request.RequestID ||
		tasks[0].Status != downloader.StatusCompleted {
		t.Fatalf("durable reconciliation tasks = %#v", tasks)
	}
}

func TestQueueRetryFailedTask(t *testing.T) {
	root := t.TempDir()
	transport := &flakyTransport{}
	observer := newEventObserver()
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     transport,
		Observer:      observer,
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	request := downloader.EnqueueRequest{
		RequestID:  "request-retry",
		AppID:      "app-retry",
		URL:        "https://download.invalid/retry",
		TotalKnown: true,
		Total:      4,
	}
	if _, err := queue.Enqueue(context.Background(), request); err != nil {
		t.Fatalf("Enqueue() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventFailed, request.RequestID)
	if err := queue.Retry(context.Background(), request.RequestID); err != nil {
		t.Fatalf("Retry() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventCompleted, request.RequestID)
	task, found := queue.Snapshot(request.RequestID)
	if !found || task.Status != downloader.StatusCompleted || task.Attempt != 2 {
		t.Fatalf("retried task = %#v, found=%t", task, found)
	}
}

func TestQueueUnknownProgressIsMonotonicThrottledAndExplicit(t *testing.T) {
	root := t.TempDir()
	observer := newEventObserver()
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     &chunkedUnknownTransport{data: []byte("data")},
		Observer:      observer,
		Clock:         newAdvancingClock(),
		MaxUnknown:    8,
		ProgressEvery: 3 * time.Second,
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	request := downloader.EnqueueRequest{
		RequestID: "request-unknown-progress",
		AppID:     "app-unknown-progress",
		URL:       "https://download.invalid/unknown-progress",
	}
	if _, err := queue.Enqueue(context.Background(), request); err != nil {
		t.Fatalf("Enqueue() error = %v", err)
	}
	waitEvent(t, observer.events, downloader.EventCompleted, request.RequestID)
	var started []downloader.Event
	var progress []downloader.Event
	for _, event := range observer.snapshot() {
		if event.RequestID != request.RequestID {
			continue
		}
		switch event.Type {
		case downloader.EventStarted:
			started = append(started, event)
		case downloader.EventProgress:
			progress = append(progress, event)
		}
	}
	if len(started) != 1 || started[0].TotalKnown || started[0].Total != 0 {
		t.Fatalf("started events = %#v", started)
	}
	if len(progress) != 2 {
		t.Fatalf("progress events = %#v, want throttled [3,4]", progress)
	}
	if progress[0].Done != 3 || progress[1].Done != 4 ||
		progress[0].Attempt != progress[1].Attempt ||
		progress[0].TotalKnown || progress[1].TotalKnown ||
		progress[0].Total != 0 || progress[1].Total != 0 {
		t.Fatalf("progress events = %#v", progress)
	}
	if progress[0].SpeedBytesSec != 1 {
		t.Fatalf("first speed = %d, want 1", progress[0].SpeedBytesSec)
	}
}

func TestQueueUnknownTransferEnforcesResponseAndResumeLimits(t *testing.T) {
	t.Run("fresh response exceeds cap", func(t *testing.T) {
		root := t.TempDir()
		observer := newEventObserver()
		queue, err := downloader.NewQueue(downloader.QueueConfig{
			DownloadsRoot: root,
			Store:         storage.NewDownloadTaskStore(root),
			Transport:     &recordingFullTransport{data: []byte("12345")},
			Observer:      observer,
			MaxUnknown:    4,
		})
		if err != nil {
			t.Fatalf("NewQueue() error = %v", err)
		}
		defer queue.Close(context.Background())
		request := downloader.EnqueueRequest{
			RequestID: "request-unknown-too-large",
			AppID:     "app-unknown-too-large",
			URL:       "https://download.invalid/unknown-too-large",
		}
		if _, err := queue.Enqueue(context.Background(), request); err != nil {
			t.Fatalf("Enqueue() error = %v", err)
		}
		failed := waitEvent(
			t,
			observer.events,
			downloader.EventFailed,
			request.RequestID,
		)
		if failed.ErrorCode != "transfer_too_large" {
			t.Fatalf("error code = %q", failed.ErrorCode)
		}
	})

	t.Run("resume requires complete total", func(t *testing.T) {
		root := t.TempDir()
		store := storage.NewDownloadTaskStore(root)
		task := downloader.Task{
			SchemaVersion: downloader.TaskSchemaVersion,
			RequestID:     "request-unknown-resume",
			AppID:         "app-unknown-resume",
			URL:           "https://download.invalid/unknown-resume",
			Status:        downloader.StatusPaused,
			Attempt:       1,
			Done:          2,
			Validator:     downloader.EntityValidator{ETag: `"v1"`},
			CreatedAt:     "2026-07-16T00:00:00Z",
		}
		if err := store.Write(task); err != nil {
			t.Fatalf("Write() error = %v", err)
		}
		paths, _ := downloader.DeriveTaskPaths(root, task.RequestID)
		if err := os.WriteFile(paths.Part, []byte("ab"), 0o600); err != nil {
			t.Fatalf("WriteFile(part) error = %v", err)
		}
		observer := newEventObserver()
		queue, err := downloader.NewQueue(downloader.QueueConfig{
			DownloadsRoot: root,
			Store:         store,
			Transport:     &unknownResumeTransport{},
			Observer:      observer,
			MaxUnknown:    8,
		})
		if err != nil {
			t.Fatalf("NewQueue() error = %v", err)
		}
		defer queue.Close(context.Background())
		if err := queue.Resume(context.Background(), task.RequestID); err != nil {
			t.Fatalf("Resume() error = %v", err)
		}
		failed := waitEvent(
			t,
			observer.events,
			downloader.EventFailed,
			task.RequestID,
		)
		if failed.ErrorCode != "transfer_incomplete" {
			t.Fatalf("error code = %q", failed.ErrorCode)
		}
	})
}

func TestQueueRecoveryUsesPartLengthAndDowngradesDownloading(t *testing.T) {
	root := t.TempDir()
	store := storage.NewDownloadTaskStore(root)
	task := downloader.Task{
		SchemaVersion: downloader.TaskSchemaVersion,
		RequestID:     "request-recover",
		AppID:         "app-recover",
		URL:           "https://download.invalid/recover",
		Status:        downloader.StatusDownloading,
		Attempt:       3,
		Done:          1,
		TotalKnown:    true,
		Total:         4,
		Validator:     downloader.EntityValidator{ETag: `"v1"`},
		CreatedAt:     "2026-07-16T00:00:00Z",
	}
	if err := store.Write(task); err != nil {
		t.Fatalf("Write() error = %v", err)
	}
	paths, _ := downloader.DeriveTaskPaths(root, task.RequestID)
	if err := os.WriteFile(paths.Part, []byte("ab"), 0o600); err != nil {
		t.Fatalf("WriteFile(part) error = %v", err)
	}
	transport := newBlockingTransport([]byte("abcd"))
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         store,
		Transport:     transport,
		Observer:      newEventObserver(),
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	open := waitOpen(t, transport.opens)
	if open.Offset != 2 || open.Validator.ETag != `"v1"` {
		t.Fatalf("recovery open = %#v", open)
	}
	if err := queue.Close(context.Background()); err != nil {
		t.Fatalf("Close() error = %v", err)
	}
	loaded, err := store.LoadAll()
	if err != nil {
		t.Fatalf("LoadAll() error = %v", err)
	}
	if len(loaded) != 1 ||
		loaded[0].Status != downloader.StatusQueued ||
		loaded[0].Done != 2 {
		t.Fatalf("recovered metadata = %#v", loaded)
	}
}

func TestQueueRecoveryFinalizesCompletePart(t *testing.T) {
	root := t.TempDir()
	store := storage.NewDownloadTaskStore(root)
	task := downloader.Task{
		SchemaVersion: downloader.TaskSchemaVersion,
		RequestID:     "request-finalize",
		AppID:         "app-finalize",
		URL:           "https://download.invalid/finalize",
		Status:        downloader.StatusQueued,
		Done:          3,
		TotalKnown:    true,
		Total:         4,
		Validator:     downloader.EntityValidator{ETag: `"v1"`},
		CreatedAt:     "2026-07-16T00:00:00Z",
	}
	if err := store.Write(task); err != nil {
		t.Fatalf("Write() error = %v", err)
	}
	paths, _ := downloader.DeriveTaskPaths(root, task.RequestID)
	if err := os.WriteFile(paths.Part, []byte("data"), 0o600); err != nil {
		t.Fatalf("WriteFile(part) error = %v", err)
	}
	transport := &countingTransport{}
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         store,
		Transport:     transport,
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	recovered, found := queue.Snapshot(task.RequestID)
	if !found || recovered.Status != downloader.StatusCompleted || recovered.Done != 4 {
		t.Fatalf("recovered task = %#v, found=%t", recovered, found)
	}
	if transport.calls != 0 {
		t.Fatalf("transport calls = %d, want 0", transport.calls)
	}
	if _, err := os.Stat(paths.Completed); err != nil {
		t.Fatalf("completed file missing: %v", err)
	}
	if _, err := os.Stat(paths.Part); !os.IsNotExist(err) {
		t.Fatalf("part still exists: %v", err)
	}
}

func TestQueueRecoveryFailsClosedOnInconsistentFiles(t *testing.T) {
	tests := []struct {
		name  string
		setup func(t *testing.T, paths downloader.TaskPaths)
	}{
		{
			name:  "completed metadata missing final",
			setup: func(t *testing.T, _ downloader.TaskPaths) {},
		},
		{
			name: "part and final both exist",
			setup: func(t *testing.T, paths downloader.TaskPaths) {
				if err := os.WriteFile(paths.Part, []byte("data"), 0o600); err != nil {
					t.Fatalf("WriteFile(part) error = %v", err)
				}
				if err := os.WriteFile(paths.Completed, []byte("data"), 0o600); err != nil {
					t.Fatalf("WriteFile(final) error = %v", err)
				}
			},
		},
	}
	for _, test := range tests {
		t.Run(test.name, func(t *testing.T) {
			root := t.TempDir()
			store := storage.NewDownloadTaskStore(root)
			task := downloader.Task{
				SchemaVersion: downloader.TaskSchemaVersion,
				RequestID:     "request-corrupt",
				AppID:         "app-corrupt",
				URL:           "https://download.invalid/corrupt",
				Status:        downloader.StatusCompleted,
				Done:          4,
				TotalKnown:    true,
				Total:         4,
				CreatedAt:     "2026-07-16T00:00:00Z",
			}
			if err := store.Write(task); err != nil {
				t.Fatalf("Write() error = %v", err)
			}
			paths, _ := downloader.DeriveTaskPaths(root, task.RequestID)
			test.setup(t, paths)
			_, err := downloader.NewQueue(downloader.QueueConfig{
				DownloadsRoot: root,
				Store:         store,
				Transport:     &countingTransport{},
			})
			if !errors.Is(err, downloader.ErrTaskCorrupt) {
				t.Fatalf("NewQueue() error = %v, want %v", err, downloader.ErrTaskCorrupt)
			}
		})
	}
}

func TestQueueRecoveryRejectsOversizedUnknownPart(t *testing.T) {
	root := t.TempDir()
	store := storage.NewDownloadTaskStore(root)
	task := downloader.Task{
		SchemaVersion: downloader.TaskSchemaVersion,
		RequestID:     "request-unknown-oversized",
		AppID:         "app-unknown-oversized",
		URL:           "https://download.invalid/unknown-oversized",
		Status:        downloader.StatusPaused,
		Done:          5,
		CreatedAt:     "2026-07-16T00:00:00Z",
	}
	if err := store.Write(task); err != nil {
		t.Fatalf("Write() error = %v", err)
	}
	paths, _ := downloader.DeriveTaskPaths(root, task.RequestID)
	if err := os.WriteFile(paths.Part, []byte("12345"), 0o600); err != nil {
		t.Fatalf("WriteFile(part) error = %v", err)
	}
	_, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         store,
		Transport:     &countingTransport{},
		MaxUnknown:    4,
	})
	if !errors.Is(err, downloader.ErrTaskCorrupt) {
		t.Fatalf("NewQueue() error = %v, want %v", err, downloader.ErrTaskCorrupt)
	}
}

func TestQueueEnqueueIdentityAndAppConflict(t *testing.T) {
	root := t.TempDir()
	queue, err := downloader.NewQueue(downloader.QueueConfig{
		DownloadsRoot: root,
		Store:         storage.NewDownloadTaskStore(root),
		Transport:     newBlockingTransport([]byte("data")),
	})
	if err != nil {
		t.Fatalf("NewQueue() error = %v", err)
	}
	defer queue.Close(context.Background())
	request := downloader.EnqueueRequest{
		RequestID:  "request-one",
		AppID:      "app-one",
		URL:        "https://download.invalid/one",
		TotalKnown: true,
		Total:      4,
	}
	if _, err := queue.Enqueue(context.Background(), request); err != nil {
		t.Fatalf("Enqueue() error = %v", err)
	}
	if _, err := queue.Enqueue(context.Background(), request); err != nil {
		t.Fatalf("idempotent Enqueue() error = %v", err)
	}
	conflict := request
	conflict.RequestID = "request-two"
	if _, err := queue.Enqueue(context.Background(), conflict); !errors.Is(err, downloader.ErrTaskConflict) {
		t.Fatalf("app conflict error = %v, want %v", err, downloader.ErrTaskConflict)
	}
}

type eventObserver struct {
	events  chan downloader.Event
	mu      sync.Mutex
	history []downloader.Event
}

type failingEventObserver struct {
	*eventObserver
	failType downloader.EventType
}

func newEventObserver() *eventObserver {
	return &eventObserver{events: make(chan downloader.Event, 128)}
}

func newFailingEventObserver(eventType downloader.EventType) *failingEventObserver {
	return &failingEventObserver{
		eventObserver: newEventObserver(),
		failType:      eventType,
	}
}

func (observer *failingEventObserver) PublishDownload(
	ctx context.Context,
	event downloader.Event,
) error {
	if err := observer.eventObserver.PublishDownload(ctx, event); err != nil {
		return err
	}
	if event.Type == observer.failType {
		return errors.New("injected observer failure")
	}
	return nil
}

func (observer *eventObserver) PublishDownload(
	ctx context.Context,
	event downloader.Event,
) error {
	observer.mu.Lock()
	observer.history = append(observer.history, event)
	observer.mu.Unlock()
	select {
	case <-ctx.Done():
		return ctx.Err()
	case observer.events <- event:
		return nil
	}
}

func (observer *eventObserver) snapshot() []downloader.Event {
	observer.mu.Lock()
	defer observer.mu.Unlock()
	return append([]downloader.Event(nil), observer.history...)
}

func waitEvent(
	t *testing.T,
	events <-chan downloader.Event,
	eventType downloader.EventType,
	requestID string,
) downloader.Event {
	t.Helper()
	timer := time.NewTimer(3 * time.Second)
	defer timer.Stop()
	for {
		select {
		case event := <-events:
			if event.Type == eventType && event.RequestID == requestID {
				return event
			}
		case <-timer.C:
			t.Fatalf("timed out waiting for %s/%s", eventType, requestID)
		}
	}
}

func assertNoEvent(
	t *testing.T,
	events <-chan downloader.Event,
	eventType downloader.EventType,
	requestID string,
) {
	t.Helper()
	timer := time.NewTimer(100 * time.Millisecond)
	defer timer.Stop()
	for {
		select {
		case event := <-events:
			if event.Type == eventType && event.RequestID == requestID {
				t.Fatalf("unexpected event %#v", event)
			}
		case <-timer.C:
			return
		}
	}
}

type blockingTransport struct {
	data []byte

	mu        sync.Mutex
	releases  map[string]chan struct{}
	active    int
	maxActive int
	opens     chan downloader.OpenRequest
}

func newBlockingTransport(data []byte) *blockingTransport {
	return &blockingTransport{
		data:     append([]byte(nil), data...),
		releases: make(map[string]chan struct{}),
		opens:    make(chan downloader.OpenRequest, 32),
	}
}

func (transport *blockingTransport) Open(
	ctx context.Context,
	request downloader.OpenRequest,
) (downloader.OpenResponse, error) {
	transport.mu.Lock()
	release := make(chan struct{})
	transport.releases[request.URL] = release
	transport.active++
	if transport.active > transport.maxActive {
		transport.maxActive = transport.active
	}
	transport.mu.Unlock()
	transport.opens <- request

	offset := request.Offset
	if offset > int64(len(transport.data)) {
		return downloader.OpenResponse{}, downloader.ErrRangeMismatch
	}
	body := &blockingBody{
		ctx:     ctx,
		release: release,
		data:    append([]byte(nil), transport.data[offset:]...),
		onClose: func() {
			transport.mu.Lock()
			transport.active--
			transport.mu.Unlock()
		},
	}
	return downloader.OpenResponse{
		Body:                body,
		TotalKnown:          true,
		Total:               int64(len(transport.data)),
		ResponseLengthKnown: true,
		ResponseLength:      int64(len(transport.data)) - offset,
		Validator:           downloader.EntityValidator{ETag: `"v1"`},
	}, nil
}

func (transport *blockingTransport) release(url string) {
	transport.mu.Lock()
	release := transport.releases[url]
	transport.mu.Unlock()
	if release != nil {
		close(release)
	}
}

func (transport *blockingTransport) maxActiveValue() int {
	transport.mu.Lock()
	defer transport.mu.Unlock()
	return transport.maxActive
}

type blockingBody struct {
	ctx     context.Context
	release <-chan struct{}
	data    []byte
	sent    bool
	onClose func()
	once    sync.Once
}

func (body *blockingBody) Read(buffer []byte) (int, error) {
	if body.sent {
		return 0, io.EOF
	}
	select {
	case <-body.ctx.Done():
		return 0, body.ctx.Err()
	case <-body.release:
	}
	body.sent = true
	return copy(buffer, body.data), nil
}

func (body *blockingBody) Close() error {
	body.once.Do(body.onClose)
	return nil
}

type resumeTransport struct {
	mu    sync.Mutex
	opens []downloader.OpenRequest
}

func (transport *resumeTransport) Open(
	ctx context.Context,
	request downloader.OpenRequest,
) (downloader.OpenResponse, error) {
	transport.mu.Lock()
	transport.opens = append(transport.opens, request)
	transport.mu.Unlock()
	switch request.Offset {
	case 0:
		return downloader.OpenResponse{
			Body:                &prefixThenBlockBody{ctx: ctx, prefix: []byte("ab")},
			TotalKnown:          true,
			Total:               4,
			ResponseLengthKnown: true,
			ResponseLength:      4,
			Validator:           downloader.EntityValidator{ETag: `"v1"`},
		}, nil
	case 2:
		return downloader.OpenResponse{
			Body:                io.NopCloser(bytes.NewReader([]byte("cd"))),
			TotalKnown:          true,
			Total:               4,
			ResponseLengthKnown: true,
			ResponseLength:      2,
			Validator:           downloader.EntityValidator{ETag: `"v1"`},
		}, nil
	default:
		return downloader.OpenResponse{}, downloader.ErrRangeMismatch
	}
}

func (transport *resumeTransport) openRequests() []downloader.OpenRequest {
	transport.mu.Lock()
	defer transport.mu.Unlock()
	return append([]downloader.OpenRequest(nil), transport.opens...)
}

type prefixThenBlockBody struct {
	ctx    context.Context
	prefix []byte
	sent   bool
}

func (body *prefixThenBlockBody) Read(buffer []byte) (int, error) {
	if !body.sent {
		body.sent = true
		return copy(buffer, body.prefix), nil
	}
	<-body.ctx.Done()
	return 0, body.ctx.Err()
}

func (body *prefixThenBlockBody) Close() error {
	return nil
}

var errInjectedBodyClose = errors.New("injected response body close failure")

type closeErrorTransport struct {
	opened chan struct{}
}

func (transport *closeErrorTransport) Open(
	ctx context.Context,
	_ downloader.OpenRequest,
) (downloader.OpenResponse, error) {
	return downloader.OpenResponse{
		Body:                &closeErrorBody{ctx: ctx, opened: transport.opened},
		TotalKnown:          true,
		Total:               4,
		ResponseLengthKnown: true,
		ResponseLength:      4,
		Validator:           downloader.EntityValidator{ETag: `"v1"`},
	}, nil
}

type closeErrorBody struct {
	ctx    context.Context
	opened chan<- struct{}
}

func (body *closeErrorBody) Read([]byte) (int, error) {
	body.opened <- struct{}{}
	<-body.ctx.Done()
	return 0, body.ctx.Err()
}

func (*closeErrorBody) Close() error {
	return errInjectedBodyClose
}

type fullThenBlockTransport struct {
	data []byte

	mu      sync.Mutex
	calls   int
	written chan struct{}
}

func newFullThenBlockTransport(data []byte) *fullThenBlockTransport {
	return &fullThenBlockTransport{
		data:    append([]byte(nil), data...),
		written: make(chan struct{}),
	}
}

func (transport *fullThenBlockTransport) Open(
	ctx context.Context,
	_ downloader.OpenRequest,
) (downloader.OpenResponse, error) {
	transport.mu.Lock()
	transport.calls++
	call := transport.calls
	transport.mu.Unlock()
	if call != 1 {
		return downloader.OpenResponse{}, errors.New("unexpected second transport open")
	}
	return downloader.OpenResponse{
		Body: &fullThenBlockBody{
			ctx:     ctx,
			data:    transport.data,
			written: transport.written,
		},
		TotalKnown:          true,
		Total:               int64(len(transport.data)),
		ResponseLengthKnown: true,
		ResponseLength:      int64(len(transport.data)),
		Validator:           downloader.EntityValidator{ETag: `"v1"`},
	}, nil
}

func (transport *fullThenBlockTransport) waitWritten(t *testing.T) {
	t.Helper()
	select {
	case <-transport.written:
	case <-time.After(3 * time.Second):
		t.Fatal("timed out waiting for complete part write")
	}
}

func (transport *fullThenBlockTransport) callsValue() int {
	transport.mu.Lock()
	defer transport.mu.Unlock()
	return transport.calls
}

type fullThenBlockBody struct {
	ctx     context.Context
	data    []byte
	written chan<- struct{}
	sent    bool
}

func (body *fullThenBlockBody) Read(buffer []byte) (int, error) {
	if !body.sent {
		body.sent = true
		body.written <- struct{}{}
		return copy(buffer, body.data), nil
	}
	<-body.ctx.Done()
	return 0, body.ctx.Err()
}

func (*fullThenBlockBody) Close() error {
	return nil
}

type fullThenErrorTransport struct {
	data []byte

	mu    sync.Mutex
	calls int
}

func (transport *fullThenErrorTransport) Open(
	_ context.Context,
	_ downloader.OpenRequest,
) (downloader.OpenResponse, error) {
	transport.mu.Lock()
	transport.calls++
	call := transport.calls
	transport.mu.Unlock()
	if call != 1 {
		return downloader.OpenResponse{}, errors.New("unexpected second transport open")
	}
	return downloader.OpenResponse{
		Body:                io.NopCloser(&errorAfterReader{data: transport.data}),
		TotalKnown:          true,
		Total:               int64(len(transport.data)),
		ResponseLengthKnown: true,
		ResponseLength:      int64(len(transport.data)),
		Validator:           downloader.EntityValidator{ETag: `"v1"`},
	}, nil
}

func (transport *fullThenErrorTransport) callsValue() int {
	transport.mu.Lock()
	defer transport.mu.Unlock()
	return transport.calls
}

type flakyTransport struct {
	mu    sync.Mutex
	calls int
}

type interruptedTransport struct {
	mu    sync.Mutex
	opens []downloader.OpenRequest
}

func (transport *interruptedTransport) Open(
	_ context.Context,
	request downloader.OpenRequest,
) (downloader.OpenResponse, error) {
	transport.mu.Lock()
	transport.opens = append(transport.opens, request)
	call := len(transport.opens)
	transport.mu.Unlock()
	if call == 1 {
		return downloader.OpenResponse{
			Body:                io.NopCloser(&errorAfterReader{data: []byte("ab")}),
			TotalKnown:          true,
			Total:               4,
			ResponseLengthKnown: true,
			ResponseLength:      4,
			Validator:           downloader.EntityValidator{ETag: `"v1"`},
		}, nil
	}
	return downloader.OpenResponse{
		Body:                io.NopCloser(bytes.NewReader([]byte("cd"))),
		TotalKnown:          true,
		Total:               4,
		ResponseLengthKnown: true,
		ResponseLength:      2,
		Validator:           downloader.EntityValidator{ETag: `"v1"`},
	}, nil
}

func (transport *interruptedTransport) openRequests() []downloader.OpenRequest {
	transport.mu.Lock()
	defer transport.mu.Unlock()
	return append([]downloader.OpenRequest(nil), transport.opens...)
}

type errorAfterReader struct {
	data []byte
	sent bool
}

func (reader *errorAfterReader) Read(buffer []byte) (int, error) {
	if !reader.sent {
		reader.sent = true
		return copy(buffer, reader.data), nil
	}
	return 0, errors.New("connection interrupted")
}

type restartTransport struct{}

func (*restartTransport) Open(
	_ context.Context,
	request downloader.OpenRequest,
) (downloader.OpenResponse, error) {
	if request.Offset != 2 || request.Validator.ETag != `"v1"` {
		return downloader.OpenResponse{}, downloader.ErrRangeMismatch
	}
	return downloader.OpenResponse{
		Body:                io.NopCloser(bytes.NewReader([]byte("WXYZ"))),
		Restart:             true,
		TotalKnown:          true,
		Total:               4,
		ResponseLengthKnown: true,
		ResponseLength:      4,
		Validator:           downloader.EntityValidator{ETag: `"v2"`},
	}, nil
}

type restartWriteFailStore struct {
	inner  *storage.DownloadTaskStore
	mu     sync.Mutex
	failed bool
}

func (store *restartWriteFailStore) Write(task downloader.Task) error {
	store.mu.Lock()
	defer store.mu.Unlock()
	if !store.failed &&
		task.Status == downloader.StatusDownloading &&
		task.Attempt == 3 {
		store.failed = true
		return errors.New("injected restart metadata write failure")
	}
	return store.inner.Write(task)
}

func (store *restartWriteFailStore) LoadAll() ([]downloader.Task, error) {
	return store.inner.LoadAll()
}

func (store *restartWriteFailStore) Delete(requestID string) error {
	return store.inner.Delete(requestID)
}

type recordingFullTransport struct {
	data []byte

	mu    sync.Mutex
	opens []downloader.OpenRequest
}

func (transport *recordingFullTransport) Open(
	_ context.Context,
	request downloader.OpenRequest,
) (downloader.OpenResponse, error) {
	transport.mu.Lock()
	transport.opens = append(transport.opens, request)
	transport.mu.Unlock()
	if request.Offset < 0 || request.Offset > int64(len(transport.data)) {
		return downloader.OpenResponse{}, downloader.ErrRangeMismatch
	}
	return downloader.OpenResponse{
		Body:                io.NopCloser(bytes.NewReader(transport.data[request.Offset:])),
		TotalKnown:          true,
		Total:               int64(len(transport.data)),
		ResponseLengthKnown: true,
		ResponseLength:      int64(len(transport.data)) - request.Offset,
		Validator:           downloader.EntityValidator{ETag: `"v1"`},
	}, nil
}

func (transport *recordingFullTransport) openRequests() []downloader.OpenRequest {
	transport.mu.Lock()
	defer transport.mu.Unlock()
	return append([]downloader.OpenRequest(nil), transport.opens...)
}

type chunkedUnknownTransport struct {
	data []byte
}

func (transport *chunkedUnknownTransport) Open(
	_ context.Context,
	request downloader.OpenRequest,
) (downloader.OpenResponse, error) {
	if request.Offset != 0 {
		return downloader.OpenResponse{}, downloader.ErrRangeMismatch
	}
	return downloader.OpenResponse{
		Body:                io.NopCloser(&oneByteReader{data: transport.data}),
		ResponseLengthKnown: true,
		ResponseLength:      int64(len(transport.data)),
		Validator:           downloader.EntityValidator{ETag: `"v1"`},
	}, nil
}

type unknownResumeTransport struct{}

func (*unknownResumeTransport) Open(
	_ context.Context,
	request downloader.OpenRequest,
) (downloader.OpenResponse, error) {
	if request.Offset != 2 || request.Validator.ETag != `"v1"` {
		return downloader.OpenResponse{}, downloader.ErrRangeMismatch
	}
	return downloader.OpenResponse{
		Body:                io.NopCloser(bytes.NewReader([]byte("cd"))),
		ResponseLengthKnown: true,
		ResponseLength:      2,
		Validator:           downloader.EntityValidator{ETag: `"v1"`},
	}, nil
}

type oneByteReader struct {
	data   []byte
	offset int
}

func (reader *oneByteReader) Read(buffer []byte) (int, error) {
	if reader.offset >= len(reader.data) {
		return 0, io.EOF
	}
	buffer[0] = reader.data[reader.offset]
	reader.offset++
	return 1, nil
}

type countingTransport struct {
	calls int
}

func (transport *countingTransport) Open(
	context.Context,
	downloader.OpenRequest,
) (downloader.OpenResponse, error) {
	transport.calls++
	return downloader.OpenResponse{}, errors.New("unexpected transport call")
}

type advancingClock struct {
	mu  sync.Mutex
	now time.Time
}

func newAdvancingClock() *advancingClock {
	return &advancingClock{
		now: time.Date(2026, 7, 16, 0, 0, 0, 0, time.UTC),
	}
}

func (clock *advancingClock) Now() time.Time {
	clock.mu.Lock()
	defer clock.mu.Unlock()
	current := clock.now
	clock.now = clock.now.Add(time.Second)
	return current
}

func (transport *flakyTransport) Open(
	context.Context,
	downloader.OpenRequest,
) (downloader.OpenResponse, error) {
	transport.mu.Lock()
	defer transport.mu.Unlock()
	transport.calls++
	if transport.calls == 1 {
		return downloader.OpenResponse{}, errors.New("temporary network failure")
	}
	return downloader.OpenResponse{
		Body:                io.NopCloser(bytes.NewReader([]byte("data"))),
		TotalKnown:          true,
		Total:               4,
		ResponseLengthKnown: true,
		ResponseLength:      4,
		Validator:           downloader.EntityValidator{ETag: `"v1"`},
	}, nil
}

func waitOpen(
	t *testing.T,
	opens <-chan downloader.OpenRequest,
) downloader.OpenRequest {
	t.Helper()
	select {
	case request := <-opens:
		return request
	case <-time.After(3 * time.Second):
		t.Fatal("timed out waiting for transfer start")
		return downloader.OpenRequest{}
	}
}