diff --git a/pkg/mediorum/server/blob_fetch.go b/pkg/mediorum/server/blob_fetch.go
new file mode 100644
index 00000000..3e86322e
--- /dev/null
+++ b/pkg/mediorum/server/blob_fetch.go
@@ -0,0 +1,231 @@
+package server
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"strconv"
+	"strings"
+
+	"github.com/OpenAudio/go-openaudio/pkg/mediorum/server/signature"
+)
+
+// blobFetchChunkAttempts is per chunk, not per blob. A failure costs one chunk
+// rather than restarting a multi-gigabyte transfer.
+const blobFetchChunkAttempts = 3
+
+// blobFetchChunkSize bounds how much one request may transfer, which is what
+// lets an ordinary client timeout do the job. peerHTTPClient allows three
+// minutes, so a chunk this size fails only below ~11 Mbit/s -- generous for a
+// bucket-to-node link, and a chunk that has not finished in that window is
+// stalled rather than merely large.
+//
+// The alternative, one request for the whole blob, is why long-form audio did
+// not replicate: a 1.9GB original needed ~85 Mbit/s sustained between one
+// specific pair of nodes for the entire transfer, or the client gave up.
+//
+// A var only so tests can shrink it; nothing reassigns it at runtime.
+var blobFetchChunkSize int64 = 256 << 20
+
+// chunkedBlobReader streams a blob from a peer as a series of ranged requests.
+//
+// It reads like any other body, so callers that io.Copy it into a bucket or a
+// temp file need no changes. What differs is that no single request is
+// unbounded, so peerHTTPClient's timeout applies to a known quantity of bytes.
+type chunkedBlobReader struct {
+	ctx  context.Context
+	ss   *MediorumServer
+	cid  string
+	host string
+
+	// origin is this peer's blob endpoint. target is where ranges are actually
+	// fetched from: the same endpoint when the peer served us bytes directly, or
+	// a presigned bucket URL when it redirected. signTarget tracks which, since
+	// only the peer endpoint wants our signature -- a presigned URL carries its
+	// own auth and Go strips ours across the redirect anyway.
+	origin     string
+	target     string
+	signTarget bool
+
+	total  int64
+	offset int64
+
+	cur io.ReadCloser
+
+	// singleStream means the peer ignored Range and handed back the whole body:
+	// an older node still using c.Stream. Consume it as one stream rather than
+	// re-requesting ranges it will not honour.
+	singleStream bool
+}
+
+func (r *chunkedBlobReader) rangeRequest(ctx context.Context, rawURL string, sign bool, start int64) (*http.Request, error) {
+	var req *http.Request
+	var err error
+	if sign {
+		req, err = signature.SignedGet(ctx, rawURL, r.ss.Config.privateKey, r.ss.Config.Self.Host)
+	} else {
+		req, err = http.NewRequestWithContext(ctx, http.MethodGet, rawURL, nil)
+	}
+	if err != nil {
+		return nil, err
+	}
+	end := start + blobFetchChunkSize - 1
+	req.Header.Set("Range", fmt.Sprintf("bytes=%d-%d", start, end))
+	return req, nil
+}
+
+// parseContentRange reads the total size out of "bytes 0-1023/4096".
+func parseContentRange(v string) (int64, error) {
+	slash := strings.LastIndex(v, "/")
+	if slash < 0 {
+		return 0, fmt.Errorf("malformed Content-Range %q", v)
+	}
+	sizePart := v[slash+1:]
+	if sizePart == "*" {
+		return 0, fmt.Errorf("Content-Range %q has unknown total size", v)
+	}
+	return strconv.ParseInt(sizePart, 10, 64)
+}
+
+// start issues the first ranged request against the peer and works out how the
+// rest of the transfer should proceed.
+//
+// The Range header rides through the redirect -- Go strips only sensitive
+// headers when a redirect crosses hosts -- so this one request either returns
+// the first chunk from the bucket or the first chunk from the peer, and
+// resp.Request.URL tells us which we got.
+func (r *chunkedBlobReader) start() error {
+	req, err := r.rangeRequest(r.ctx, r.origin, true, 0)
+	if err != nil {
+		return err
+	}
+	resp, err := r.ss.peerHTTPClient.Do(req)
+	if err != nil {
+		return err
+	}
+
+	switch resp.StatusCode {
+	case http.StatusPartialContent:
+		total, err := parseContentRange(resp.Header.Get("Content-Range"))
+		if err != nil {
+			resp.Body.Close()
+			return err
+		}
+		r.total = total
+		final := resp.Request.URL.String()
+		r.target = final
+		r.signTarget = final == r.origin
+		r.cur = resp.Body
+		return nil
+
+	case http.StatusOK:
+		// Range ignored: an older peer on c.Stream, or a body small enough that
+		// the server chose not to partial it. Either way there is nothing to
+		// chunk.
+		r.singleStream = true
+		r.cur = resp.Body
+		return nil
+
+	default:
+		resp.Body.Close()
+		return fmt.Errorf("pull blob: bad status: %d cid: %s host: %s", resp.StatusCode, r.cid, r.host)
+	}
+}
+
+// nextChunk fetches the range beginning at r.offset.
+func (r *chunkedBlobReader) nextChunk() error {
+	var lastErr error
+	for attempt := 0; attempt < blobFetchChunkAttempts; attempt++ {
+		req, err := r.rangeRequest(r.ctx, r.target, r.signTarget, r.offset)
+		if err != nil {
+			return err
+		}
+		resp, err := r.ss.peerHTTPClient.Do(req)
+		if err != nil {
+			lastErr = err
+			continue
+		}
+		if resp.StatusCode == http.StatusPartialContent {
+			r.cur = resp.Body
+			return nil
+		}
+		resp.Body.Close()
+
+		// A presigned URL has a finite life and a large blob can outlast it.
+		// Ask the peer again and carry on from the same offset.
+		if !r.signTarget && (resp.StatusCode == http.StatusForbidden || resp.StatusCode == http.StatusUnauthorized) {
+			if err := r.refreshTarget(); err != nil {
+				return err
+			}
+			continue
+		}
+		lastErr = fmt.Errorf("pull blob chunk at %d: bad status: %d cid: %s host: %s",
+			r.offset, resp.StatusCode, r.cid, r.host)
+	}
+	if lastErr == nil {
+		lastErr = fmt.Errorf("pull blob chunk at %d: exhausted attempts cid: %s host: %s", r.offset, r.cid, r.host)
+	}
+	return lastErr
+}
+
+// refreshTarget re-resolves an expired presigned URL by asking the peer for a
+// zero-length range, which costs one round trip and no payload.
+func (r *chunkedBlobReader) refreshTarget() error {
+	req, err := signature.SignedGet(r.ctx, r.origin, r.ss.Config.privateKey, r.ss.Config.Self.Host)
+	if err != nil {
+		return err
+	}
+	req.Header.Set("Range", "bytes=0-0")
+	resp, err := r.ss.peerHTTPClient.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+	if resp.StatusCode != http.StatusPartialContent && resp.StatusCode != http.StatusOK {
+		return fmt.Errorf("refresh blob target: bad status: %d cid: %s host: %s", resp.StatusCode, r.cid, r.host)
+	}
+	io.Copy(io.Discard, resp.Body)
+	final := resp.Request.URL.String()
+	r.target = final
+	r.signTarget = final == r.origin
+	return nil
+}
+
+func (r *chunkedBlobReader) Read(p []byte) (int, error) {
+	for {
+		if r.cur == nil {
+			return 0, io.EOF
+		}
+		n, err := r.cur.Read(p)
+		if n > 0 {
+			r.offset += int64(n)
+			return n, nil
+		}
+		if err == nil {
+			continue
+		}
+		if !errors.Is(err, io.EOF) {
+			return 0, err
+		}
+
+		r.cur.Close()
+		r.cur = nil
+		if r.singleStream || r.offset >= r.total {
+			return 0, io.EOF
+		}
+		if err := r.nextChunk(); err != nil {
+			return 0, err
+		}
+	}
+}
+
+func (r *chunkedBlobReader) Close() error {
+	if r.cur != nil {
+		err := r.cur.Close()
+		r.cur = nil
+		return err
+	}
+	return nil
+}
diff --git a/pkg/mediorum/server/blob_fetch_test.go b/pkg/mediorum/server/blob_fetch_test.go
new file mode 100644
index 00000000..1256c60d
--- /dev/null
+++ b/pkg/mediorum/server/blob_fetch_test.go
@@ -0,0 +1,185 @@
+package server
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/OpenAudio/go-openaudio/pkg/registrar"
+	"github.com/stretchr/testify/require"
+	"go.uber.org/zap"
+)
+
+// blobFetchTestServer builds the minimum a chunkedBlobReader touches: a signing
+// key, a self host, and the peer client whose timeout bounds each chunk.
+func blobFetchTestServer(t *testing.T) *MediorumServer {
+	t.Helper()
+	return &MediorumServer{
+		Config: MediorumConfig{
+			Self:       registrar.Peer{Host: "http://self.test"},
+			privateKey: generateTestPrivateKey(1),
+		},
+		peerHTTPClient: &http.Client{Timeout: 10 * time.Second},
+		logger:         zap.NewNop(),
+	}
+}
+
+// rangeServer serves body honouring Range, counting requests.
+func rangeServer(body []byte, requests *atomic.Int64) *httptest.Server {
+	return httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if requests != nil {
+			requests.Add(1)
+		}
+		http.ServeContent(w, r, "blob", time.Unix(0, 0), bytes.NewReader(body))
+	}))
+}
+
+func readAllFrom(t *testing.T, ss *MediorumServer, origin string) ([]byte, error) {
+	t.Helper()
+	r := &chunkedBlobReader{ctx: context.Background(), ss: ss, cid: "testcid", host: "peer", origin: origin}
+	if err := r.start(); err != nil {
+		return nil, err
+	}
+	defer r.Close()
+	return io.ReadAll(r)
+}
+
+// The point of chunking: no single request covers the whole blob, so an
+// ordinary client timeout bounds a known quantity of bytes.
+func TestChunkedBlobReaderFetchesInRanges(t *testing.T) {
+	ss := blobFetchTestServer(t)
+
+	// Three chunks and a remainder, without allocating 256MB per chunk.
+	body := bytes.Repeat([]byte("mediorum"), 4096)
+	var requests atomic.Int64
+	srv := rangeServer(body, &requests)
+	defer srv.Close()
+
+	withChunkSize(t, len(body)/3)
+
+	got, err := readAllFrom(t, ss, srv.URL)
+	require.NoError(t, err)
+	require.True(t, bytes.Equal(body, got), "assembled blob differs from the source")
+	require.Greater(t, requests.Load(), int64(1), "fetched in a single request; chunking did not happen")
+}
+
+// An older peer still on c.Stream ignores Range and returns the whole body.
+// The reader must consume that rather than re-requesting ranges it won't honour.
+func TestChunkedBlobReaderFallsBackWhenRangeIgnored(t *testing.T) {
+	ss := blobFetchTestServer(t)
+	body := bytes.Repeat([]byte("x"), 5000)
+
+	var requests atomic.Int64
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		requests.Add(1)
+		w.WriteHeader(http.StatusOK)
+		w.Write(body)
+	}))
+	defer srv.Close()
+
+	withChunkSize(t, 1000)
+
+	got, err := readAllFrom(t, ss, srv.URL)
+	require.NoError(t, err)
+	require.True(t, bytes.Equal(body, got))
+	require.EqualValues(t, 1, requests.Load(), "kept issuing ranges to a peer that ignores them")
+}
+
+// A presigned URL has a finite life and a large blob can outlast it. Expiry
+// mid-transfer must re-resolve and carry on from the same offset, not restart.
+func TestChunkedBlobReaderRefreshesExpiredPresignedURL(t *testing.T) {
+	ss := blobFetchTestServer(t)
+	body := bytes.Repeat([]byte("abcdefgh"), 2048)
+
+	var bucketHits atomic.Int64
+	var expired atomic.Bool
+	expired.Store(true)
+
+	bucket := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Fail exactly once, partway in, the way a lapsed signature would.
+		if bucketHits.Add(1) == 2 && expired.CompareAndSwap(true, false) {
+			w.WriteHeader(http.StatusForbidden)
+			return
+		}
+		http.ServeContent(w, r, "blob", time.Unix(0, 0), bytes.NewReader(body))
+	}))
+	defer bucket.Close()
+
+	origin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		http.Redirect(w, r, bucket.URL, http.StatusTemporaryRedirect)
+	}))
+	defer origin.Close()
+
+	withChunkSize(t, len(body)/4)
+
+	got, err := readAllFrom(t, ss, origin.URL)
+	require.NoError(t, err)
+	require.True(t, bytes.Equal(body, got), "resumed transfer does not match the source")
+	require.False(t, expired.Load(), "the expiry branch never ran; the test proved nothing")
+}
+
+// A chunk failure costs one chunk, not the whole blob.
+func TestChunkedBlobReaderRetriesOneChunk(t *testing.T) {
+	ss := blobFetchTestServer(t)
+	body := bytes.Repeat([]byte("0123456789"), 1024)
+
+	var failed atomic.Bool
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		// Fail the first non-initial range once.
+		if strings.HasPrefix(r.Header.Get("Range"), "bytes=") &&
+			!strings.HasPrefix(r.Header.Get("Range"), "bytes=0-") &&
+			failed.CompareAndSwap(false, true) {
+			w.WriteHeader(http.StatusInternalServerError)
+			return
+		}
+		http.ServeContent(w, r, "blob", time.Unix(0, 0), bytes.NewReader(body))
+	}))
+	defer srv.Close()
+
+	withChunkSize(t, len(body)/3)
+
+	got, err := readAllFrom(t, ss, srv.URL)
+	require.NoError(t, err)
+	require.True(t, bytes.Equal(body, got))
+	require.True(t, failed.Load(), "the retry branch never ran; the test proved nothing")
+}
+
+func TestChunkedBlobReaderSurfacesBadStatus(t *testing.T) {
+	ss := blobFetchTestServer(t)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusNotFound)
+	}))
+	defer srv.Close()
+
+	_, err := readAllFrom(t, ss, srv.URL)
+	require.Error(t, err)
+	require.Contains(t, err.Error(), fmt.Sprint(http.StatusNotFound))
+}
+
+func TestParseContentRange(t *testing.T) {
+	total, err := parseContentRange("bytes 0-1023/4096")
+	require.NoError(t, err)
+	require.EqualValues(t, 4096, total)
+
+	_, err = parseContentRange("bytes 0-1023/*")
+	require.Error(t, err, "unknown total size must not be read as a length")
+
+	_, err = parseContentRange("nonsense")
+	require.Error(t, err)
+}
+
+// withChunkSize shrinks the chunk for the duration of a test so a few kilobytes
+// exercise the same multi-request path a multi-gigabyte blob would.
+func withChunkSize(t *testing.T, n int) {
+	t.Helper()
+	original := blobFetchChunkSize
+	blobFetchChunkSize = int64(n)
+	t.Cleanup(func() { blobFetchChunkSize = original })
+}
diff --git a/pkg/mediorum/server/blob_pull_async.go b/pkg/mediorum/server/blob_pull_async.go
new file mode 100644
index 00000000..9331d325
--- /dev/null
+++ b/pkg/mediorum/server/blob_pull_async.go
@@ -0,0 +1,126 @@
+package server
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+
+	"go.uber.org/zap"
+)
+
+const (
+	// asyncPullWorkers bounds how many transfers this node runs at once.
+	//
+	// This is the backpressure that a synchronous pull used to provide by
+	// accident: the sender held one of its own workers for the duration, so no
+	// node could have more transfers in flight than the sender had workers. A
+	// sender that returns immediately can queue as fast as it enumerates, so the
+	// limit has to live here, on the side actually moving the bytes.
+	asyncPullWorkers = 3
+
+	// asyncPullQueueDepth is deliberately shallow. A deep queue would accept
+	// work this node cannot start for a long time, and the sender would have
+	// stopped waiting for it -- 503 tells the sender to come back on its next
+	// sweep instead, when the picture may have changed.
+	asyncPullQueueDepth = 32
+
+	// asyncPullTimeout bounds one queued transfer. The request context cannot be
+	// used: it is cancelled the moment the handler returns 202.
+	asyncPullTimeout = 60 * time.Minute
+)
+
+// errAsyncPullQueueFull is answered with 503, which senders treat as a plain
+// failure. It must not look like "pull unsupported", or the sender falls back
+// to pushing the bytes at a node that just said it was busy.
+var errAsyncPullQueueFull = errors.New("pull queue is full")
+
+type asyncPullJob struct {
+	sourceHost     string
+	cid            string
+	placementHosts []string
+	uploadID       string
+	transcoded     bool
+}
+
+// enqueueAsyncPull accepts a transfer to run in the background, or reports why
+// it will not.
+//
+// Deduplication matters more here than it did synchronously. The sender's sweep,
+// other senders, and repair can all ask for the same cid, and previously the
+// combination of haveInMyBucket and a blocked caller kept that to one transfer
+// at a time. Nothing blocks now, so the in-flight set is what prevents the same
+// blob being fetched several times over.
+func (ss *MediorumServer) enqueueAsyncPull(job asyncPullJob) error {
+	ss.asyncPullMu.Lock()
+	if _, running := ss.asyncPullInFlight[job.cid]; running {
+		ss.asyncPullMu.Unlock()
+		return nil
+	}
+	ss.asyncPullInFlight[job.cid] = struct{}{}
+	ss.asyncPullMu.Unlock()
+
+	select {
+	case ss.asyncPullQueue <- job:
+		return nil
+	default:
+		ss.releaseAsyncPull(job.cid)
+		return errAsyncPullQueueFull
+	}
+}
+
+func (ss *MediorumServer) releaseAsyncPull(cid string) {
+	ss.asyncPullMu.Lock()
+	delete(ss.asyncPullInFlight, cid)
+	ss.asyncPullMu.Unlock()
+}
+
+func (ss *MediorumServer) startAsyncPullWorkers(ctx context.Context) error {
+	ss.logger.Info("starting async blob pull workers", zap.Int("count", asyncPullWorkers))
+
+	var wg sync.WaitGroup
+	for range asyncPullWorkers {
+		wg.Add(1)
+		go func() {
+			defer wg.Done()
+			ss.asyncPullWorker(ctx)
+		}()
+	}
+	wg.Wait()
+	return nil
+}
+
+func (ss *MediorumServer) asyncPullWorker(ctx context.Context) {
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case job := <-ss.asyncPullQueue:
+			ss.runAsyncPull(ctx, job)
+		}
+	}
+}
+
+func (ss *MediorumServer) runAsyncPull(parent context.Context, job asyncPullJob) {
+	defer ss.releaseAsyncPull(job.cid)
+
+	// Deliberately not the request context, which died with the 202 response.
+	// Parented to the server's lifecycle so shutdown still cancels in-flight
+	// transfers rather than leaking them.
+	ctx, cancel := context.WithTimeout(parent, asyncPullTimeout)
+	defer cancel()
+
+	err := ss.pullFileFromHostValidated(ctx, job.sourceHost, job.cid, job.placementHosts, job.uploadID, job.transcoded)
+	if err != nil {
+		// Nothing is waiting on this, so a log is the only report. The sender
+		// finds out on its next sweep, when the peer answers something other
+		// than already_present.
+		ss.logger.Warn("async blob pull failed",
+			zap.String("sourceHost", job.sourceHost),
+			zap.String("cid", job.cid),
+			zap.Error(err),
+		)
+		return
+	}
+	ss.logger.Debug("async blob pull complete", zap.String("cid", job.cid))
+}
diff --git a/pkg/mediorum/server/blob_pull_async_test.go b/pkg/mediorum/server/blob_pull_async_test.go
new file mode 100644
index 00000000..2172e934
--- /dev/null
+++ b/pkg/mediorum/server/blob_pull_async_test.go
@@ -0,0 +1,166 @@
+package server
+
+import (
+	"context"
+	"errors"
+	"net/http"
+	"net/http/httptest"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/require"
+)
+
+func asyncPullTestServer(t *testing.T, depth int) *MediorumServer {
+	t.Helper()
+	ss := blobFetchTestServer(t)
+	ss.asyncPullQueue = make(chan asyncPullJob, depth)
+	ss.asyncPullInFlight = map[string]struct{}{}
+	return ss
+}
+
+func TestEnqueueAsyncPullAccepts(t *testing.T) {
+	ss := asyncPullTestServer(t, 4)
+	require.NoError(t, ss.enqueueAsyncPull(asyncPullJob{cid: "cid-a", sourceHost: "http://peer"}))
+	require.Len(t, ss.asyncPullQueue, 1)
+}
+
+// Nothing blocks the sender any more, so the sweep, other senders and repair can
+// all ask for the same cid at once. Previously haveInMyBucket plus a blocked
+// caller kept that to one transfer; now the in-flight set has to.
+func TestEnqueueAsyncPullDeduplicatesByCID(t *testing.T) {
+	ss := asyncPullTestServer(t, 8)
+
+	for range 5 {
+		require.NoError(t, ss.enqueueAsyncPull(asyncPullJob{cid: "same-cid", sourceHost: "http://peer"}))
+	}
+	require.Len(t, ss.asyncPullQueue, 1, "queued the same blob more than once")
+
+	// A different cid is unaffected.
+	require.NoError(t, ss.enqueueAsyncPull(asyncPullJob{cid: "other-cid", sourceHost: "http://peer"}))
+	require.Len(t, ss.asyncPullQueue, 2)
+}
+
+// A full queue must report busy rather than accept work it cannot start, and
+// must release the in-flight marker so a later sweep can retry.
+func TestEnqueueAsyncPullRejectsWhenFull(t *testing.T) {
+	ss := asyncPullTestServer(t, 1)
+
+	require.NoError(t, ss.enqueueAsyncPull(asyncPullJob{cid: "first", sourceHost: "http://peer"}))
+	err := ss.enqueueAsyncPull(asyncPullJob{cid: "second", sourceHost: "http://peer"})
+	require.ErrorIs(t, err, errAsyncPullQueueFull)
+
+	ss.asyncPullMu.Lock()
+	_, stillMarked := ss.asyncPullInFlight["second"]
+	ss.asyncPullMu.Unlock()
+	require.False(t, stillMarked, "a rejected job stayed marked in flight and could never be retried")
+}
+
+// Completion must clear the marker, or that cid can never be pulled again for
+// the life of the process.
+func TestAsyncPullReleasesInFlightMarker(t *testing.T) {
+	ss := asyncPullTestServer(t, 2)
+	require.NoError(t, ss.enqueueAsyncPull(asyncPullJob{cid: "cid-x", sourceHost: "http://peer"}))
+
+	ss.releaseAsyncPull("cid-x")
+
+	require.NoError(t, ss.enqueueAsyncPull(asyncPullJob{cid: "cid-x", sourceHost: "http://peer"}),
+		"cid could not be re-queued after release")
+	require.Len(t, ss.asyncPullQueue, 2)
+}
+
+// The trap this refactor invites: c.Request().Context() is cancelled the moment
+// the handler returns 202, so a background job holding it would be killed
+// immediately. runAsyncPull must derive its own.
+func TestAsyncPullDoesNotInheritACancelledRequestContext(t *testing.T) {
+	ss := asyncPullTestServer(t, 1)
+
+	// Stand in for the request context: already cancelled, as it would be by
+	// the time a queued job ran.
+	requestCtx, cancel := context.WithCancel(context.Background())
+	cancel()
+
+	// The worker is parented to the server lifecycle, not the request.
+	serverCtx := context.Background()
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		// A failing pull is fine: what matters is that it was attempted rather
+		// than short-circuited by a dead context.
+		ss.runAsyncPull(serverCtx, asyncPullJob{cid: "ctx-cid", sourceHost: "http://127.0.0.1:1"})
+	}()
+
+	select {
+	case <-done:
+	case <-time.After(30 * time.Second):
+		t.Fatal("async pull did not finish")
+	}
+
+	require.Error(t, requestCtx.Err(), "sanity: the stand-in request context should be cancelled")
+
+	ss.asyncPullMu.Lock()
+	_, stillMarked := ss.asyncPullInFlight["ctx-cid"]
+	ss.asyncPullMu.Unlock()
+	require.False(t, stillMarked, "in-flight marker leaked after the job finished")
+}
+
+// The sender's reaction to each answer a receiver can give. What matters is not
+// the error text but whether it routes into the multipart push fallback: a peer
+// that is busy or already fetching must never be sent the bytes.
+func TestRequestPeerPullStatusHandling(t *testing.T) {
+	cases := []struct {
+		name          string
+		status        int
+		wantErr       error
+		wantFallback  bool
+		wantInProress bool
+	}{
+		{name: "already present", status: http.StatusOK},
+		{name: "accepted for async pull", status: http.StatusAccepted, wantErr: errPeerPullInProgress, wantInProress: true},
+		{name: "queue full", status: http.StatusServiceUnavailable},
+		{name: "endpoint absent", status: http.StatusNotFound, wantErr: errPeerPullUnsupported, wantFallback: true},
+		{name: "not implemented", status: http.StatusNotImplemented, wantErr: errPeerPullUnsupported, wantFallback: true},
+		{name: "peer gateway failure", status: http.StatusBadGateway, wantErr: errPeerPullFailed, wantFallback: true},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			ss := asyncPullTestServer(t, 1)
+			srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+				w.WriteHeader(tc.status)
+			}))
+			defer srv.Close()
+
+			err := ss.requestPeerPull(context.Background(), srv.URL, "cid", nil, "", false)
+
+			if tc.status == http.StatusOK {
+				require.NoError(t, err)
+				return
+			}
+			require.Error(t, err)
+			if tc.wantErr != nil {
+				require.ErrorIs(t, err, tc.wantErr)
+			}
+			require.Equal(t, tc.wantFallback, isPullFallbackWorthy(err),
+				"wrong fallback decision for %d: pushing bytes at a peer that did not ask for them", tc.status)
+			require.Equal(t, tc.wantInProress, errors.Is(err, errPeerPullInProgress))
+		})
+	}
+}
+
+// 503 is the receiver saying it has no room to work. Treating it as
+// "pull unsupported" would answer that by pushing the whole blob at it.
+func TestQueueFullDoesNotTriggerMultipartFallback(t *testing.T) {
+	ss := asyncPullTestServer(t, 1)
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		w.WriteHeader(http.StatusServiceUnavailable)
+		w.Write([]byte(`{"error":"` + errAsyncPullQueueFull.Error() + `"}`))
+	}))
+	defer srv.Close()
+
+	err := ss.requestPeerPull(context.Background(), srv.URL, "cid", nil, "", false)
+	require.Error(t, err)
+	require.False(t, isPullFallbackWorthy(err),
+		"a busy peer would be sent the bytes over multipart")
+}
diff --git a/pkg/mediorum/server/replicate.go b/pkg/mediorum/server/replicate.go
index aa04e928..fe1c3f18 100644
--- a/pkg/mediorum/server/replicate.go
+++ b/pkg/mediorum/server/replicate.go
@@ -29,6 +29,12 @@ var (
 	errPeerPullUnsupported   = errors.New("peer does not support pull replication")
 	errPeerPullFailed        = errors.New("peer could not pull blob")
 	errPulledBlobCIDMismatch = errors.New("pulled blob CID mismatch")
+	// errPeerPullInProgress means the peer accepted the transfer and is running
+	// it in the background. Not a success -- there is no mirror yet -- and not a
+	// failure, so it must not be logged as one, and above all must not fall
+	// through to the multipart push, which would send the peer the very bytes it
+	// is already fetching.
+	errPeerPullInProgress = errors.New("peer accepted pull; transfer in progress")
 )
 
 const maxPeerErrorBytes = 8 << 10
@@ -273,7 +279,7 @@ func (ss *MediorumServer) replicateStoredFileToHost(
 		if err == nil {
 			return nil
 		}
-		if !errors.Is(err, errPeerPullUnsupported) && !errors.Is(err, errPeerPullFailed) {
+		if !isPullFallbackWorthy(err) {
 			return err
 		}
 		ss.logger.Debug("falling back to multipart blob replication",
@@ -292,12 +298,24 @@ func (ss *MediorumServer) replicateStoredFileToHost(
 	return ss.replicateFileToHost(ctx, peer, fileName, reader, placementHosts)
 }
 
+// isPullFallbackWorthy reports whether a failed pull should be retried by
+// pushing the bytes over multipart instead.
+//
+// Only two answers qualify: the peer cannot do pull at all, or it tried and
+// failed. Everything else must not be sent bytes -- a 503 means the peer is out
+// of room to work, and pushing at it is the worst possible response; a 202
+// means it is already fetching them itself.
+func isPullFallbackWorthy(err error) bool {
+	return errors.Is(err, errPeerPullUnsupported) || errors.Is(err, errPeerPullFailed)
+}
+
 func (ss *MediorumServer) requestPeerPull(ctx context.Context, peer, cid string, placementHosts []string, uploadID string, transcoded bool) error {
 	payload, err := json.Marshal(internalBlobPullRequest{
 		CID:            cid,
 		PlacementHosts: placementHosts,
 		UploadID:       uploadID,
 		Transcoded:     transcoded,
+		Async:          true,
 	})
 	if err != nil {
 		return err
@@ -325,6 +343,8 @@ func (ss *MediorumServer) requestPeerPull(ctx context.Context, peer, cid string,
 	case http.StatusOK:
 		_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxPeerErrorBytes))
 		return nil
+	case http.StatusAccepted:
+		return errPeerPullInProgress
 	case http.StatusNotFound, http.StatusMethodNotAllowed, http.StatusNotImplemented:
 		_, _ = io.Copy(io.Discard, io.LimitReader(resp.Body, maxPeerErrorBytes))
 		return fmt.Errorf("%w: %s", errPeerPullUnsupported, resp.Status)
@@ -459,27 +479,25 @@ func (ss *MediorumServer) openBlobFromHost(ctx context.Context, host, cid string
 	if host == ss.Config.Self.Host {
 		return nil, errors.New("should not pull blob from self")
 	}
-	u := apiPath(host, "internal/blobs", url.PathEscape(cid))
-
-	req, err := signature.SignedGet(ctx, u, ss.Config.privateKey, ss.Config.Self.Host)
-	if err != nil {
-		return nil, err
-	}
 	// The peer's GET endpoint uses hot-first-then-archive fallback, so it
 	// finds the blob without placement context. Placement only governs the
 	// receiver's local write after this stream is validated.
-
-	resp, err := ss.peerHTTPClient.Do(req)
-	if err != nil {
+	//
+	// The body arrives as a series of ranged requests rather than one open
+	// stream, so peerHTTPClient's timeout bounds a known quantity of bytes
+	// instead of a whole transfer of unknown size. Callers see an ordinary
+	// ReadCloser either way.
+	r := &chunkedBlobReader{
+		ctx:    ctx,
+		ss:     ss,
+		cid:    cid,
+		host:   host,
+		origin: apiPath(host, "internal/blobs", url.PathEscape(cid)),
+	}
+	if err := r.start(); err != nil {
 		return nil, err
 	}
-
-	if resp.StatusCode != 200 {
-		resp.Body.Close()
-		return nil, fmt.Errorf("pull blob: bad status: %d cid: %s host: %s", resp.StatusCode, cid, host)
-	}
-
-	return resp.Body, nil
+	return r, nil
 }
 
 // diskWarnInterval caps how often each dsnHasSpace warn is emitted per DSN.
diff --git a/pkg/mediorum/server/replicate_test.go b/pkg/mediorum/server/replicate_test.go
index ae748598..200efd71 100644
--- a/pkg/mediorum/server/replicate_test.go
+++ b/pkg/mediorum/server/replicate_test.go
@@ -69,6 +69,26 @@ func TestReplicateFileToHostClosesPipeOnEarlyHTTPError(t *testing.T) {
 	assert.LessOrEqual(t, countReplicateFileToHostGoroutines(), before)
 }
 
+// The sender now hands the transfer off and returns immediately, so both of
+// these assert against the receiver's settled state rather than the sender's
+// error. waitForAsyncPull makes that deterministic: the in-flight marker is set
+// before the handler answers 202 and cleared when the job finishes, so there is
+// no window where the test could observe an unstarted transfer.
+func waitForAsyncPull(t *testing.T, ss *MediorumServer, cid string) {
+	t.Helper()
+	deadline := time.Now().Add(30 * time.Second)
+	for time.Now().Before(deadline) {
+		ss.asyncPullMu.Lock()
+		_, running := ss.asyncPullInFlight[cid]
+		ss.asyncPullMu.Unlock()
+		if !running {
+			return
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatalf("async pull of %s did not finish", cid)
+}
+
 func TestRequestPeerPullStoresValidatedBlob(t *testing.T) {
 	source := testNetwork[0]
 	target := testNetwork[1]
@@ -90,7 +110,10 @@ func TestRequestPeerPullStoresValidatedBlob(t *testing.T) {
 		"",
 		true,
 	)
-	require.NoError(t, err)
+	// Accepted, not complete: the bytes have not moved yet.
+	require.ErrorIs(t, err, errPeerPullInProgress)
+
+	waitForAsyncPull(t, target, cid)
 
 	reader, _, err := target.readBlob(context.Background(), cidutil.ShardCID(cid))
 	require.NoError(t, err)
@@ -100,6 +123,11 @@ func TestRequestPeerPullStoresValidatedBlob(t *testing.T) {
 	require.Equal(t, content, string(stored))
 }
 
+// Validation still happens on the receiver; what changed is that its verdict is
+// no longer reported to the sender. The guarantee that matters -- a blob whose
+// content does not match its CID is never stored -- is unaffected, and the
+// sender learns of the failure by the peer not answering already_present on the
+// next sweep.
 func TestRequestPeerPullRejectsCIDMismatch(t *testing.T) {
 	source := testNetwork[0]
 	target := testNetwork[1]
@@ -120,9 +148,10 @@ func TestRequestPeerPullRejectsCIDMismatch(t *testing.T) {
 		"",
 		true,
 	)
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "422")
-	require.False(t, target.haveInMyBucket(cid))
+	require.ErrorIs(t, err, errPeerPullInProgress)
+
+	waitForAsyncPull(t, target, cid)
+	require.False(t, target.haveInMyBucket(cid), "stored a blob that failed CID validation")
 }
 
 func TestReplicateStoredFileToHostUsesPullWithoutReadingSourceBucket(t *testing.T) {
diff --git a/pkg/mediorum/server/replicate_worker.go b/pkg/mediorum/server/replicate_worker.go
index 1db8bac3..04e75541 100644
--- a/pkg/mediorum/server/replicate_worker.go
+++ b/pkg/mediorum/server/replicate_worker.go
@@ -2,6 +2,7 @@ package server
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"sync"
 	"time"
@@ -171,6 +172,17 @@ func (ss *MediorumServer) replicateToHosts(ctx context.Context, upload *Upload,
 	// Collect results
 	newSuccessHosts := []string{}
 	for result := range resultsChan {
+		if errors.Is(result.err, errPeerPullInProgress) {
+			// The peer owns this transfer now. Recording a mirror would be a
+			// claim we cannot back, and logging a failure would be wrong too.
+			// The next sweep asks again and gets already_present -- the peer
+			// reporting what is actually in its bucket, which is a better
+			// signal than anything it could have promised us here.
+			ss.logger.Debug("peer accepted blob pull; awaiting confirmation on a later sweep",
+				zap.String("host", result.host),
+				zap.String("cid", cid))
+			continue
+		}
 		if result.err != nil {
 			fileType := "file"
 			if isTranscoded {
diff --git a/pkg/mediorum/server/serve_blob.go b/pkg/mediorum/server/serve_blob.go
index b2cb085b..66a05209 100644
--- a/pkg/mediorum/server/serve_blob.go
+++ b/pkg/mediorum/server/serve_blob.go
@@ -610,7 +610,18 @@ func (ss *MediorumServer) serveInternalBlobGET(c echo.Context) error {
 	}
 	defer blob.Close()
 
-	return c.Stream(200, blob.ContentType(), blob)
+	// ServeContent rather than c.Stream: it answers Range requests with 206 and
+	// a Content-Range, which is what lets a peer fetch this blob in bounded
+	// chunks. c.Stream ignores Range and always returns the whole body, so a
+	// backend that cannot presign -- file://, which is a supported production
+	// storage driver, not only a dev convenience -- would otherwise force the
+	// receiver back onto a single unbounded stream.
+	//
+	// Content-Type is set explicitly because ServeContent would otherwise sniff
+	// it or guess from the name, and a bare cid carries no extension.
+	c.Response().Header().Set(echo.HeaderContentType, blob.ContentType())
+	http.ServeContent(c.Response(), c.Request(), cid, blob.ModTime(), blob)
+	return nil
 }
 
 type internalBlobPullRequest struct {
@@ -628,6 +639,14 @@ type internalBlobPullRequest struct {
 	// analysis targets -- decoding them costs an ffmpeg subprocess each to
 	// produce a waveform for a cid nothing will ever ask about.
 	Transcoded bool `json:"transcoded,omitempty"`
+	// Async asks this node to accept the transfer and run it in the background,
+	// answering 202 rather than holding the sender open for its duration.
+	//
+	// Opt-in, and that is what makes a rolling deploy uneventful. A sender that
+	// predates the field sends nothing and gets the synchronous path it expects;
+	// a peer that predates it ignores the flag and also stays synchronous. The
+	// sender only stops waiting when both ends understand the handoff.
+	Async bool `json:"async,omitempty"`
 }
 
 func (ss *MediorumServer) serveInternalBlobPull(c echo.Context) error {
@@ -657,6 +676,23 @@ func (ss *MediorumServer) serveInternalBlobPull(c echo.Context) error {
 		return c.JSON(http.StatusServiceUnavailable, map[string]string{"error": "disk is too full to accept new blobs"})
 	}
 
+	if request.Async {
+		job := asyncPullJob{
+			sourceHost:     sourceHost,
+			cid:            request.CID,
+			placementHosts: request.PlacementHosts,
+			uploadID:       request.UploadID,
+			transcoded:     request.Transcoded,
+		}
+		if err := ss.enqueueAsyncPull(job); err != nil {
+			// Busy, not incapable. 503 keeps the sender off the multipart
+			// fallback, which would push the bytes at a node that just said it
+			// had no room to work.
+			return c.JSON(http.StatusServiceUnavailable, map[string]string{"error": err.Error()})
+		}
+		return c.JSON(http.StatusAccepted, map[string]string{"status": "accepted"})
+	}
+
 	err := ss.pullFileFromHostValidated(c.Request().Context(), sourceHost, request.CID, request.PlacementHosts, request.UploadID, request.Transcoded)
 	if errors.Is(err, errPulledBlobCIDMismatch) {
 		return c.JSON(http.StatusUnprocessableEntity, map[string]string{"error": err.Error()})
diff --git a/pkg/mediorum/server/server.go b/pkg/mediorum/server/server.go
index b94aac08..cb3e1dd6 100644
--- a/pkg/mediorum/server/server.go
+++ b/pkg/mediorum/server/server.go
@@ -138,8 +138,14 @@ type MediorumServer struct {
 	rendezvousHasher *common.RendezvousHasher
 	transcodeWork    chan *Upload
 	replicationWork  chan *Upload
-	waveformWork     chan waveformJob
-	ethService       ethv1connect.EthServiceHandler
+
+	// Bounded queue for transfers a peer handed off with 202. The limit lives
+	// here because the sender no longer blocks for the duration.
+	asyncPullQueue    chan asyncPullJob
+	asyncPullInFlight map[string]struct{}
+	asyncPullMu       sync.Mutex
+	waveformWork      chan waveformJob
+	ethService        ethv1connect.EthServiceHandler
 
 	// isDbLocalhost is reported by the health check. Derived once from the
 	// parsed connection config below, since the DSN can't change at runtime.
@@ -456,10 +462,12 @@ func New(lc *lifecycle.Lifecycle, logger *zap.Logger, config MediorumConfig, pos
 		// Buffered, unlike the audio-analysis channel: the sweeps must not
 		// block on a full queue, and the route enqueues opportunistically and
 		// gives up rather than waiting.
-		waveformWork:    make(chan waveformJob, 256),
-		replicationWork: make(chan *Upload, 100),
-		posChannel:      posChannel,
-		pruneTrigger:    make(chan pruneRequest, 1),
+		waveformWork:      make(chan waveformJob, 256),
+		replicationWork:   make(chan *Upload, 100),
+		asyncPullQueue:    make(chan asyncPullJob, asyncPullQueueDepth),
+		asyncPullInFlight: map[string]struct{}{},
+		posChannel:        posChannel,
+		pruneTrigger:      make(chan pruneRequest, 1),
 
 		peerHealths:           map[string]*PeerHealth{},
 		redirectCache:         imcache.New(imcache.WithMaxEntriesLimitOption[string, string](50_000, imcache.EvictionPolicyLRU)),
@@ -674,6 +682,7 @@ func (ss *MediorumServer) MustStart() error {
 		ss.lc.AddManagedRoutine("waveform analyzer", ss.startWaveformAnalyzer)
 	}
 	ss.lc.AddManagedRoutine("replication workers", ss.startReplicationWorkers)
+	ss.lc.AddManagedRoutine("async blob pull workers", ss.startAsyncPullWorkers)
 
 	ss.lc.AddManagedRoutine("pruner", ss.startPruner)