diff --git a/weed/operation/upload_content.go b/weed/operation/upload_content.go index 704f70f49d..081d295e66 100644 --- a/weed/operation/upload_content.go +++ b/weed/operation/upload_content.go @@ -17,6 +17,7 @@ import ( "strconv" "strings" "sync" + "sync/atomic" "time" "github.com/seaweedfs/seaweedfs/weed/util/request_id" @@ -63,6 +64,10 @@ type UploadOption struct { MaxAttempts int // <=0 uses the default GenUploadUrl func(host, fileId string) string // if nil → fallback "http://{host}/{fileId}" FilerJwt security.FilerJwtProvider // credential for proxy chunk URLs; nil → process-wide jwt.filer_signing + + // stallWatchdogDisarmed turns the watchdog off for this upload's later + // attempts once it fired: a re-arm could burn every retry. + stallWatchdogDisarmed bool } type UploadResult struct { @@ -125,6 +130,7 @@ var ( // StatusCode 0 means the request never got a response — a transport failure. type uploadStatusError struct { StatusCode int + stalled bool // the stall watchdog cancelled this attempt's request err error } @@ -163,6 +169,11 @@ type HTTPClient interface { // Uploader type Uploader struct { httpClient HTTPClient + // fastUploadHosts records the wire rate of the last sizeable upload to + // each host, so a later upload stalling far below it can be blamed on + // that one connection rather than the link and get cancelled for a retry + // on a fresh conn. + fastUploadHosts sync.Map // host:port → fastHostMark } func NewUploader() (*Uploader, error) { @@ -328,6 +339,11 @@ func (uploader *Uploader) doUpload(ctx context.Context, reader io.Reader, option } func (uploader *Uploader) retriedUploadData(ctx context.Context, data []byte, option *UploadOption) (uploadResult *UploadResult, err error) { + // A stall disarms the watchdog for this upload's remaining attempts only; + // keep that on a per-call option so a caller reusing UploadOption starts + // the next upload armed. + localOption := *option + option = &localOption maxAttempts := option.MaxAttempts if maxAttempts <= 0 { maxAttempts = 3 @@ -348,6 +364,10 @@ func (uploader *Uploader) retriedUploadData(ctx context.Context, data []byte, op if ctx.Err() != nil { return nil, ctx.Err() } + var statusErr *uploadStatusError + if errors.As(err, &statusErr) && statusErr.stalled { + option.stallWatchdogDisarmed = true + } glog.WarningfCtx(ctx, "uploading %d to %s: %v", i, option.UploadUrl, err) } return @@ -415,14 +435,15 @@ func (uploader *Uploader) doUploadData(ctx context.Context, data []byte, option _, err = w.Write(encryptedData) return }, len(encryptedData), &UploadOption{ - UploadUrl: option.UploadUrl, - Filename: "", - Cipher: false, - IsInputCompressed: false, - MimeType: "", - PairMap: nil, - Jwt: option.Jwt, - BytesBuffer: option.BytesBuffer, + UploadUrl: option.UploadUrl, + Filename: "", + Cipher: false, + IsInputCompressed: false, + MimeType: "", + PairMap: nil, + Jwt: option.Jwt, + BytesBuffer: option.BytesBuffer, + stallWatchdogDisarmed: option.stallWatchdogDisarmed, }) if uploadResult == nil { return @@ -443,15 +464,16 @@ func (uploader *Uploader) doUploadData(ctx context.Context, data []byte, option _, err = w.Write(data) return }, len(data), &UploadOption{ - UploadUrl: option.UploadUrl, - Filename: option.Filename, - Cipher: false, - IsInputCompressed: contentIsGzipped, - MimeType: option.MimeType, - PairMap: option.PairMap, - Jwt: option.Jwt, - Md5: option.Md5, - BytesBuffer: option.BytesBuffer, + UploadUrl: option.UploadUrl, + Filename: option.Filename, + Cipher: false, + IsInputCompressed: contentIsGzipped, + MimeType: option.MimeType, + PairMap: option.PairMap, + Jwt: option.Jwt, + Md5: option.Md5, + BytesBuffer: option.BytesBuffer, + stallWatchdogDisarmed: option.stallWatchdogDisarmed, }) if uploadResult == nil { return @@ -514,7 +536,14 @@ func (uploader *Uploader) upload_content(ctx context.Context, fillBufferFunction } else { reqReader = bytes.NewReader(option.BytesBuffer.Bytes()) } - req, postErr := http.NewRequestWithContext(ctx, http.MethodPost, option.UploadUrl, reqReader) + // bodySize is what the conn must move: originalDataSize counts bytes + // before compression, which would inflate both the recorded rate and the + // stall budget for a pre-compressed body. + bodySize := int64(reqReader.Len()) + reqCtx, cancelUpload := context.WithCancel(ctx) + defer cancelUpload() + + req, postErr := http.NewRequestWithContext(reqCtx, http.MethodPost, option.UploadUrl, reqReader) if postErr != nil { glog.V(1).InfofCtx(ctx, "create upload request %s: %v", option.UploadUrl, postErr) return nil, fmt.Errorf("create upload request %s: %v", option.UploadUrl, postErr) @@ -529,7 +558,46 @@ func (uploader *Uploader) upload_content(ctx context.Context, fillBufferFunction request_id.InjectToRequest(ctx, req) + var uploadDone atomic.Bool + defer uploadDone.Store(true) + // bodySentAt marks when the transport finished draining the request body; + // past it the wait is server-side commit, which the transport's response + // header timeout already bounds — the stall watchdog only blames a conn + // that is still trickling the body. + var curBodySentAt atomic.Pointer[atomic.Int64] + bodySentAt := &atomic.Int64{} + curBodySentAt.Store(bodySentAt) + req.Body = &watchdogBody{ReadCloser: req.Body, sentAt: bodySentAt} + mark, hostWasFast := uploader.fastUploadHosts.Load(req.URL.Host) + hostWasFast = hostWasFast && !option.stallWatchdogDisarmed && time.Since(time.Unix(0, mark.(fastHostMark).at)) < fastUploadHostMemory + stallFired := func() { + if !uploadDone.CompareAndSwap(false, true) { + return + } + // The host was moving bodies at full rate a moment ago, so an + // upload still trickling far below that rate rides a wedged pooled + // conn; cancel drops the conn so the retry dials a fresh one. + // CompareAndDelete leaves a newer mark a concurrent success stored. + uploader.fastUploadHosts.CompareAndDelete(req.URL.Host, mark) + glog.WarningfCtx(ctx, "upload %d bytes to %s stalled on a pooled connection, retrying on a fresh one", originalDataSize, option.UploadUrl) + cancelUpload() + } + var stallTimer *time.Timer + if hostWasFast { + defer func() { + if stallTimer != nil { + stallTimer.Stop() + } + }() + stallTimer = time.AfterFunc(uploadStallTimeout(bodySize, mark.(fastHostMark).bps), func() { + if curBodySentAt.Load().Load() == 0 { + stallFired() + } + }) + } + // print("+") + started := time.Now() resp, post_err := uploader.httpClient.Do(req) defer util_http.CloseResponse(resp) if post_err != nil { @@ -548,7 +616,13 @@ func (uploader *Uploader) upload_content(ctx context.Context, fillBufferFunction // "connection reset" error with a misleading "Body length 0". if req.GetBody != nil { if newBody, gbErr := req.GetBody(); gbErr == nil { - req.Body = newBody + bodySentAt = &atomic.Int64{} + curBodySentAt.Store(bodySentAt) + started = time.Now() + if stallTimer != nil && !uploadDone.Load() { + stallTimer.Reset(uploadStallTimeout(bodySize, mark.(fastHostMark).bps)) + } + req.Body = &watchdogBody{ReadCloser: newBody, sentAt: bodySentAt} resp, post_err = uploader.httpClient.Do(req) defer util_http.CloseResponse(resp) } else { @@ -561,22 +635,40 @@ func (uploader *Uploader) upload_content(ctx context.Context, fillBufferFunction } if post_err != nil { stats.UploadErrorCounter.WithLabelValues("0").Inc() - return nil, &uploadStatusError{StatusCode: 0, err: fmt.Errorf("upload %s %d bytes to %v: %v", option.Filename, originalDataSize, option.UploadUrl, post_err)} + return nil, &uploadStatusError{StatusCode: 0, stalled: reqCtx.Err() != nil && ctx.Err() == nil, err: fmt.Errorf("upload %s %d bytes to %v: %v", option.Filename, originalDataSize, option.UploadUrl, post_err)} } + // Rate is bounded by elapsed-to-response: the server cannot answer before + // the body lands, so this can only understate link speed — a drain-time + // reading mistakes kernel-buffer absorption for wire speed and would mark + // a slow link fast. The mark still only lands on a verified upload: a + // fast rejection says nothing about how the conn drains. + movedIn := time.Since(started) + // print("-") var ret UploadResult etag := getEtag(resp) if resp.StatusCode == http.StatusNoContent { + if bps := uploadObservedBps(bodySize, movedIn); bps > 0 { + uploader.fastUploadHosts.Store(req.URL.Host, fastHostMark{at: time.Now().UnixNano(), bps: bps}) + } ret.ETag = etag ret.ContentMd5 = resp.Header.Get("Content-MD5") return &ret, nil } + if stallTimer != nil { + // The header timeout does not cover the response body: keep the + // watchdog armed for the read so a conn that wedges mid-reply gets + // cancelled for a fresh retry. + respStallTimer := time.AfterFunc(uploadStallMinWait, stallFired) + defer respStallTimer.Stop() + } resp_body, ra_err := io.ReadAll(resp.Body) + uploadDone.Store(true) if ra_err != nil { stats.UploadErrorCounter.WithLabelValues(strconv.Itoa(resp.StatusCode)).Inc() - return nil, &uploadStatusError{StatusCode: resp.StatusCode, err: fmt.Errorf("read response body %v: %w", option.UploadUrl, ra_err)} + return nil, &uploadStatusError{StatusCode: resp.StatusCode, stalled: reqCtx.Err() != nil && ctx.Err() == nil, err: fmt.Errorf("read response body %v: %w", option.UploadUrl, ra_err)} } unmarshal_err := json.Unmarshal(resp_body, &ret) @@ -589,6 +681,9 @@ func (uploader *Uploader) upload_content(ctx context.Context, fillBufferFunction stats.UploadErrorCounter.WithLabelValues(strconv.Itoa(resp.StatusCode)).Inc() return nil, &uploadStatusError{StatusCode: resp.StatusCode, err: fmt.Errorf("unmarshalled error %v: %v", option.UploadUrl, ret.Error)} } + if bps := uploadObservedBps(bodySize, movedIn); bps > 0 { + uploader.fastUploadHosts.Store(req.URL.Host, fastHostMark{at: time.Now().UnixNano(), bps: bps}) + } ret.ETag = etag ret.ContentMd5 = resp.Header.Get("Content-MD5") return &ret, nil @@ -601,3 +696,68 @@ func getEtag(r *http.Response) (etag string) { } return } + +// uploadRateMinBytes is the smallest body whose measured rate can mark a +// host: smaller uploads are dominated by round-trip and commit overhead, not +// wire speed. +const uploadRateMinBytes = int64(256 << 10) + +var ( + // uploadStallSlowFactor is how far below the host's last observed rate an + // upload may run before its conn is presumed wedged: at 8, a host that + // just moved a body at 8MiB/s tolerates down to 1MiB/s. + uploadStallSlowFactor = int64(8) + uploadStallMinWait = 10 * time.Second + fastUploadHostMemory = time.Minute +) + +// fastHostMark remembers the wire rate of a host's last sizeable upload, so a +// later upload running far below it blames the pooled conn rather than the +// link — no fixed rate fits every deployment's links. +type fastHostMark struct { + at int64 // UnixNano + bps int64 // wire bytes/sec measured elapsed-to-response +} + +// uploadStallTimeout is the most an upload may take on a conn whose host last +// proved bps: the time the body needs at an uploadStallSlowFactor of that +// rate, with a floor so a fast conn is not raced on small bodies. +func uploadStallTimeout(size, bps int64) time.Duration { + if bps > 0 { + if d := time.Duration(size/bps) * time.Duration(uploadStallSlowFactor) * time.Second; d > uploadStallMinWait { + return d + } + } + return uploadStallMinWait +} + +// uploadObservedBps returns the wire rate a completed upload proved on its +// host, or 0 when the body was too small to measure. Elapsed-to-response can +// only understate the link's speed, which keeps the mark — and the stall +// budget derived from it — conservative. +func uploadObservedBps(size int64, elapsed time.Duration) int64 { + if size < uploadRateMinBytes || elapsed <= 0 { + return 0 + } + return int64(float64(size) / elapsed.Seconds()) +} + +// watchdogBody records when the transport drains the request body: before that +// a stuck send blames the conn; after it the wait is the server's. +type watchdogBody struct { + io.ReadCloser + sentAt *atomic.Int64 +} + +func (b *watchdogBody) Read(p []byte) (int, error) { + n, err := b.ReadCloser.Read(p) + if err == io.EOF { + b.sentAt.Store(time.Now().UnixNano()) + } + return n, err +} + +func (b *watchdogBody) Close() error { + b.sentAt.Store(time.Now().UnixNano()) + return b.ReadCloser.Close() +} diff --git a/weed/operation/upload_content_test.go b/weed/operation/upload_content_test.go index ebbfb722f4..d80c8cd591 100644 --- a/weed/operation/upload_content_test.go +++ b/weed/operation/upload_content_test.go @@ -8,10 +8,12 @@ import ( "errors" "fmt" "io" + "net" "net/http" "net/http/httptest" "strings" "sync" + "sync/atomic" "testing" "time" @@ -199,6 +201,372 @@ func (c *bodyCapturingHTTPClient) Do(req *http.Request) (*http.Response, error) }, nil } +// stallingHTTPClient parks the first stalls calls without draining the request +// body (a conn that never drains) until the request ctx is cancelled, and +// delays the rest by slowFor after draining. +type stallingHTTPClient struct { + mu sync.Mutex + stalls int + slowFor time.Duration + requestCtxs []context.Context +} + +func (c *stallingHTTPClient) Do(req *http.Request) (*http.Response, error) { + c.mu.Lock() + c.requestCtxs = append(c.requestCtxs, req.Context()) + stall := c.stalls > 0 + if stall { + c.stalls-- + } + c.mu.Unlock() + + if stall { + <-req.Context().Done() + return nil, req.Context().Err() + } + if req.Body != nil { + _, _ = io.Copy(io.Discard, req.Body) + } + if c.slowFor > 0 { + time.Sleep(c.slowFor) + } + return &http.Response{ + StatusCode: http.StatusCreated, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(`{"size":1}`)), + }, nil +} + +// markHostFast arms the stall watchdog the way a recent fast upload would: +// the recorded rate is high enough that the adaptive budget lands on +// uploadStallMinWait for test-sized bodies. +func markHostFast(u *Uploader, host string) { + u.fastUploadHosts.Store(host, fastHostMark{at: time.Now().UnixNano(), bps: 1 << 30}) +} + +// A host that just served a fast upload arms the stall watchdog: a request +// parked well past the floor is riding a wedged conn, so it is cancelled and +// the retry runs on a fresh conn without the watchdog. +func TestUploadStallWatchdogRetriesOnFreshConn(t *testing.T) { + defer func(orig time.Duration) { uploadStallMinWait = orig }(uploadStallMinWait) + uploadStallMinWait = 50 * time.Millisecond + + client := &stallingHTTPClient{stalls: 1} + uploader := newUploader(client) + markHostFast(uploader, "test-host:8080") + + result, err := uploader.UploadData(context.Background(), make([]byte, 300*1024), &UploadOption{ + UploadUrl: "http://test-host:8080/1,deadbeef", + MaxAttempts: 2, + }) + if err != nil { + t.Fatalf("upload should have retried on a fresh conn: %v", err) + } + if result.RetryCount != 1 || len(client.requestCtxs) != 2 { + t.Fatalf("got retries=%d calls=%d, want one retry on a fresh conn", result.RetryCount, len(client.requestCtxs)) + } + if err := client.requestCtxs[0].Err(); err == nil { + t.Fatalf("first attempt context was not cancelled") + } +} + +// readStallHTTPClient answers with a response body that never delivers until +// the request ctx is cancelled — the conn that took the request then wedged on +// the read side. +type readStallHTTPClient struct { + mu sync.Mutex + stall bool +} + +func (c *readStallHTTPClient) Do(req *http.Request) (*http.Response, error) { + if req.Body != nil { + _, _ = io.Copy(io.Discard, req.Body) + } + c.mu.Lock() + stall := c.stall + c.stall = false + c.mu.Unlock() + var body io.ReadCloser = io.NopCloser(strings.NewReader(`{"size":1}`)) + if stall { + body = &ctxBlockedBody{ctx: req.Context()} + } + return &http.Response{ + StatusCode: http.StatusCreated, + Header: make(http.Header), + Body: body, + }, nil +} + +type ctxBlockedBody struct{ ctx context.Context } + +func (b *ctxBlockedBody) Read([]byte) (int, error) { + <-b.ctx.Done() + return 0, b.ctx.Err() +} + +func (b *ctxBlockedBody) Close() error { return nil } + +// The watchdog stays armed through the response read: a conn that accepts the +// body then wedges on the reply is as wedged as one that never drains. +func TestUploadStallWatchdogCoversResponseRead(t *testing.T) { + defer func(orig time.Duration) { uploadStallMinWait = orig }(uploadStallMinWait) + uploadStallMinWait = 50 * time.Millisecond + + client := &readStallHTTPClient{stall: true} + uploader := newUploader(client) + markHostFast(uploader, "test-host:8080") + + result, err := uploader.UploadData(context.Background(), make([]byte, 300*1024), &UploadOption{ + UploadUrl: "http://test-host:8080/1,deadbeef", + MaxAttempts: 2, + }) + if err != nil { + t.Fatalf("upload should have retried past the stalled read: %v", err) + } + if result.RetryCount != 1 { + t.Fatalf("got retries=%d, want one retry on a fresh conn", result.RetryCount) + } +} + +// rearmHTTPClient parks the first attempt on a conn that never drains and +// re-marks the host fast as it fails — a concurrent healthy upload can restore +// the mark while this upload is between attempts. The retry drains slowly, so +// a re-armed watchdog would cancel it too. +type rearmHTTPClient struct { + uploader *Uploader + calls int32 + slowFor time.Duration +} + +func (c *rearmHTTPClient) Do(req *http.Request) (*http.Response, error) { + if atomic.AddInt32(&c.calls, 1) == 1 { + <-req.Context().Done() + markHostFast(c.uploader, req.URL.Host) + return nil, req.Context().Err() + } + // Park past the watchdog floor like a real wedged conn: a re-armed + // watchdog must cancel this attempt instead of the sleep masking it. + select { + case <-req.Context().Done(): + return nil, req.Context().Err() + case <-time.After(c.slowFor): + } + if req.Body != nil { + _, _ = io.Copy(io.Discard, req.Body) + } + return &http.Response{ + StatusCode: http.StatusCreated, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(`{"size":1}`)), + }, nil +} + +// A re-proven host must not re-arm the watchdog for the retry: the conn it +// cancelled says nothing about the next conn, and re-arming could burn every +// remaining attempt. +func TestUploadStallWatchdogStaysDisarmedOnRetry(t *testing.T) { + defer func(orig time.Duration) { uploadStallMinWait = orig }(uploadStallMinWait) + uploadStallMinWait = 40 * time.Millisecond + + uploader := newUploader(nil) + uploader.httpClient = &rearmHTTPClient{uploader: uploader, slowFor: 60 * time.Millisecond} + markHostFast(uploader, "test-host:8080") + + result, err := uploader.UploadData(context.Background(), make([]byte, 300*1024), &UploadOption{ + UploadUrl: "http://test-host:8080/1,deadbeef", + MaxAttempts: 2, + }) + if err != nil { + t.Fatalf("retry should run unbounded once the watchdog has fired: %v", err) + } + if result.RetryCount != 1 { + t.Fatalf("got retries=%d, want one retry", result.RetryCount) + } +} + +// The disarm is per upload: a caller reusing its UploadOption still gets an +// armed watchdog on the next call — a second stalled conn must cancel too. +func TestUploadStallWatchdogRearmsForNextUpload(t *testing.T) { + defer func(orig time.Duration) { uploadStallMinWait = orig }(uploadStallMinWait) + uploadStallMinWait = 40 * time.Millisecond + + client := &stallingHTTPClient{} + uploader := newUploader(client) + markHostFast(uploader, "test-host:8080") + + ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + // IsInputCompressed keeps the body large enough to re-prove the host: + // compressed near-zeros would send too few bytes to re-mark it. + option := &UploadOption{UploadUrl: "http://test-host:8080/1,deadbeef", MaxAttempts: 2, IsInputCompressed: true} + for i := 0; i < 2; i++ { + client.mu.Lock() + client.stalls = 1 + client.mu.Unlock() + result, err := uploader.UploadData(ctx, make([]byte, 300*1024), option) + if err != nil { + t.Fatalf("upload %d should have retried on a fresh conn: %v", i, err) + } + if result.RetryCount != 1 { + t.Fatalf("upload %d got retries=%d, want its own stall caught", i, result.RetryCount) + } + } + if option.stallWatchdogDisarmed { + t.Fatalf("a stall must not disarm the watchdog on the caller's option") + } +} + +// A real transport must drop the conn a cancelled request was riding: the +// retry has to dial again rather than land back on the parked conn. The first +// conn here wedges mid-reply — headers flush, then the body never arrives. +func TestUploadStallWatchdogDialsFreshConn(t *testing.T) { + defer func(orig time.Duration) { uploadStallMinWait = orig }(uploadStallMinWait) + uploadStallMinWait = 50 * time.Millisecond + + var requests, dials int32 + srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + _, _ = io.Copy(io.Discard, r.Body) + if atomic.AddInt32(&requests, 1) == 1 { + w.WriteHeader(http.StatusCreated) + w.(http.Flusher).Flush() + <-r.Context().Done() + return + } + w.WriteHeader(http.StatusCreated) + _, _ = w.Write([]byte(`{"size":1}`)) + })) + defer srv.Close() + + tr := &http.Transport{ + DialContext: func(ctx context.Context, network, addr string) (net.Conn, error) { + atomic.AddInt32(&dials, 1) + return new(net.Dialer).DialContext(ctx, network, addr) + }, + } + defer tr.CloseIdleConnections() + uploader := newUploader(&http.Client{Transport: tr}) + + host := strings.TrimPrefix(srv.URL, "http://") + markHostFast(uploader, host) + + result, err := uploader.UploadData(context.Background(), make([]byte, 300*1024), &UploadOption{ + UploadUrl: srv.URL + "/1,deadbeef", + MaxAttempts: 2, + }) + if err != nil { + t.Fatalf("upload should have retried on a fresh conn: %v", err) + } + if result.RetryCount != 1 || atomic.LoadInt32(&dials) != 2 { + t.Fatalf("got retries=%d dials=%d, want one retry on a fresh conn", result.RetryCount, dials) + } +} + +// Once the body drains the wait is the server's commit: a write that takes +// longer than the stall floor to acknowledge must not be cancelled — the conn +// is healthy, the server is just busy. +func TestUploadStallWatchdogAllowsSlowCommit(t *testing.T) { + defer func(orig time.Duration) { uploadStallMinWait = orig }(uploadStallMinWait) + uploadStallMinWait = 30 * time.Millisecond + + client := &stallingHTTPClient{slowFor: 80 * time.Millisecond} + uploader := newUploader(client) + markHostFast(uploader, "test-host:8080") + + result, err := uploader.UploadData(context.Background(), make([]byte, 300*1024), &UploadOption{ + UploadUrl: "http://test-host:8080/1,deadbeef", + MaxAttempts: 1, + }) + if err != nil || result.RetryCount != 0 { + t.Fatalf("a slow commit must not be cancelled: result=%+v err=%v", result, err) + } +} + +// A fast rejection proves nothing about how the conn drains, so it must not +// mark the host fast and arm the watchdog for the next upload. +func TestUploadStallWatchdogIgnoresFailedWrites(t *testing.T) { + uploader := newUploader(&rejectHTTPClient{}) + _, err := uploader.UploadData(context.Background(), make([]byte, 300*1024), &UploadOption{ + UploadUrl: "http://test-host:8080/1,deadbeef", + MaxAttempts: 1, + }) + if err == nil { + t.Fatalf("rejected upload should fail") + } + if _, ok := uploader.fastUploadHosts.Load("test-host:8080"); ok { + t.Fatalf("a failed upload must not mark the host fast") + } +} + +type rejectHTTPClient struct{} + +func (c *rejectHTTPClient) Do(req *http.Request) (*http.Response, error) { + if req.Body != nil { + _, _ = io.Copy(io.Discard, req.Body) + } + return &http.Response{ + StatusCode: http.StatusInternalServerError, + Header: make(http.Header), + Body: io.NopCloser(strings.NewReader(`{"error":"volume is read only"}`)), + }, nil +} + +// Without a recent fast upload the watchdog stays off: a chronically slow link +// is the link, not a conn, and must not get its uploads cancelled. +func TestUploadStallWatchdogKeepsUnprovenHost(t *testing.T) { + defer func(orig time.Duration) { uploadStallMinWait = orig }(uploadStallMinWait) + uploadStallMinWait = 20 * time.Millisecond + + client := &stallingHTTPClient{slowFor: 60 * time.Millisecond} + uploader := newUploader(client) + + result, err := uploader.UploadData(context.Background(), make([]byte, 300*1024), &UploadOption{ + UploadUrl: "http://slow-host:8080/1,deadbeef", + MaxAttempts: 1, + }) + if err != nil { + t.Fatalf("unproven host upload should not be cancelled: %v", err) + } + if result.RetryCount != 0 || len(client.requestCtxs) != 1 { + t.Fatalf("got retries=%d calls=%d, want a single unbounded attempt", result.RetryCount, len(client.requestCtxs)) + } +} + +// Any sizeable completed upload marks its host at the rate it proved — slow +// or fast — because the stall budget scales off that rate. Only a too-small +// body fails to measure. +func TestUploadObservedBps(t *testing.T) { + if bps := uploadObservedBps(4<<20, 4*time.Second); bps != 1<<20 { + t.Fatalf("4MiB in 4s observed %d B/s, want 1MiB/s", bps) + } + if bps := uploadObservedBps(4<<20, 3*time.Hour); bps <= 0 { + t.Fatalf("a slow upload still proves the rate it moved at") + } + if bps := uploadObservedBps(64*1024, time.Millisecond); bps != 0 { + t.Fatalf("a small upload says nothing about host speed") + } + if bps := uploadObservedBps(4<<20, 0); bps != 0 { + t.Fatalf("a zero-duration sample is not a measurement") + } +} + +// The stall budget scales with the rate the host last proved, so the same +// body gets little slack on a fast link and plenty on a slow one — no fixed +// floor fits both. +func TestUploadStallTimeoutScalesWithMeasuredRate(t *testing.T) { + // 8MiB/s observed → a 64MiB body gets 8x the time it needs at that rate. + if d := uploadStallTimeout(64<<20, 8<<20); d != 64*time.Second { + t.Fatalf("stall budget = %v, want 64s", d) + } + // The same body on a 64MiB/s host gets 8s — under the floor, so clamped. + if d := uploadStallTimeout(64<<20, 64<<20); d != uploadStallMinWait { + t.Fatalf("stall budget = %v, want the min wait %v", d, uploadStallMinWait) + } + // A tiny body on a fast host cannot race its own conn. + if d := uploadStallTimeout(1, 1<<30); d != uploadStallMinWait { + t.Fatalf("stall budget = %v, want the min wait %v", d, uploadStallMinWait) + } +} + type fsyncAssignClient struct { filer_pb.SeaweedFilerClient response *filer_pb.AssignVolumeResponse