From bfec34e2c3ac4984acf29d26f791d48c515a896b Mon Sep 17 00:00:00 2001 From: Marcus Pasell <3690498+rickyrombo@users.noreply.github.com> Date: Thu, 20 Aug 2026 01:26:56 -0700 Subject: [PATCH] fix(storage): replicate long blobs in bounded chunks, off the sender's worker Long-form audio did not replicate. peerHTTPClient sets a 3 minute Timeout, and http.Client.Timeout covers the body read, so it bounds how long a transfer may take rather than whether it is progressing. One request carrying a whole blob therefore demanded a sustained rate between one specific pair of nodes for its entire duration -- roughly 85 Mbit/s for a multi-gigabyte original. It failed quietly: with no successful host, replicateToHosts suppresses the write, so nothing surfaced but warn logs while the sweep re-sent three minutes of bytes every cycle and threw them away. Bound the unit of transfer and the standard mechanism becomes correct. chunkedBlobReader fetches a blob as a series of 256 MB ranged requests on the existing client, so a chunk that has not finished in three minutes is stalled rather than merely large, and a retry costs one chunk instead of the whole blob. It presents as an ordinary ReadCloser, so validation and the bucket write are untouched. No progress tracking, no bespoke stall detection, no separate client. The internal GET's non-presigning fallback moves from c.Stream to http.ServeContent so it answers Range at all. That is not a dev-only concern: file:// is a supported production storage driver, and without it those operators stay on a single unbounded stream. Second, the sender no longer waits. requestPeerPull held a replication worker for the peer's entire transfer while observing nothing, so a peer that accepted the connection and then wedged parked that worker -- its TCP stack keeps answering keepalives, so nothing detected it. The peer now answers 202 and runs the transfer on its own bounded pool. That pool is deliberate: the sender's blocking wait was accidental backpressure, capping in-flight transfers at the sender's worker count, so the limit has to be relocated to the side actually moving the bytes. A full queue answers 503, which stays out of the multipart fallback -- pushing a blob at a node that just said it had no room is the worst available response. The handoff is opt-in per request, which is what makes the rolling deploy uneventful in both directions: an old sender gets the synchronous path it expects, and an old peer ignores the flag and stays synchronous. No completion callback. already_present on the next sweep is a verified signal derived from the peer's actual bucket state, where a callback would trade evidence for an assertion and add a lost-message failure mode. Until that confirmation arrives, 202 is recorded as neither success nor failure. Upload completeness is unaffected: transcode.go marks an upload done after attestUploadCids with no replication gate, and the client returns on status done without inspecting mirrors. --- pkg/mediorum/server/blob_fetch.go | 231 ++++++++++++++++++++ pkg/mediorum/server/blob_fetch_test.go | 185 ++++++++++++++++ pkg/mediorum/server/blob_pull_async.go | 126 +++++++++++ pkg/mediorum/server/blob_pull_async_test.go | 166 ++++++++++++++ pkg/mediorum/server/replicate.go | 52 +++-- pkg/mediorum/server/replicate_test.go | 37 +++- pkg/mediorum/server/replicate_worker.go | 12 + pkg/mediorum/server/serve_blob.go | 38 +++- pkg/mediorum/server/server.go | 21 +- 9 files changed, 840 insertions(+), 28 deletions(-) create mode 100644 pkg/mediorum/server/blob_fetch.go create mode 100644 pkg/mediorum/server/blob_fetch_test.go create mode 100644 pkg/mediorum/server/blob_pull_async.go create mode 100644 pkg/mediorum/server/blob_pull_async_test.go 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)