From 11da958d11fba93f50fa88f974634e5fbfd9b34c Mon Sep 17 00:00:00 2001 From: aZuiZui <110262485+aZuiZui@users.noreply.github.com> Date: Fri, 13 Nov 2026 17:23:30 +0000 Subject: [PATCH] fix(stream): answer 502 when Telegram fails before the first byte When Telegram can't serve a part (-503 Timeout after the retries, or its message is gone), the stream had already sent 206 and its headers, so the client got an empty body cut short and had to guess. rclone retried it as a short read over and over. - Read the first 64 KiB before writing the status line. If that fails, answer 502 (telegram_part_unavailable / telegram_read_failed) and log api.stream_unreadable. Errors with their own status (504 etc.) keep it. - The same for a part whose size is looked up on open (older uploads) and is gone: it was a 500. - The retry middleware returned a bare "context deadline exceeded" when the deadline ran out while it retried a -503, so the reason was lost. It now returns an error that reads as the Telegram error and still matches both (errors.Is deadline, tgerr.As). - Part errors name the message id. --- internal/api/raw.go | 56 +++++++++- internal/api/raw_stream_test.go | 153 +++++++++++++++++++++++++++ internal/telegramstore/retry.go | 29 ++++- internal/telegramstore/retry_test.go | 39 +++++++ internal/transfer/download.go | 4 +- 5 files changed, 275 insertions(+), 6 deletions(-) create mode 100644 internal/api/raw_stream_test.go diff --git a/internal/api/raw.go b/internal/api/raw.go index 78f86322..c4518810 100644 --- a/internal/api/raw.go +++ b/internal/api/raw.go @@ -12,11 +12,13 @@ import ( "syscall" "github.com/google/uuid" + "github.com/gotd/td/tgerr" "github.com/tgdrive/teldrive/v2/internal/api/gen" "github.com/tgdrive/teldrive/v2/internal/db/sqlcgen" "github.com/tgdrive/teldrive/v2/internal/dbtypes" "github.com/tgdrive/teldrive/v2/internal/shares" + "github.com/tgdrive/teldrive/v2/internal/telegramstore" "github.com/tgdrive/teldrive/v2/internal/transfer" ) @@ -143,10 +145,30 @@ func (h *RawHandler) streamFile(ctx context.Context, w http.ResponseWriter, user UserID: userID, FileID: fileID, Offset: rangeSpec.Offset, Length: rangeSpec.Length, }) if err != nil { + if isTelegramPartGone(err) { + // Opening asks Telegram for part sizes older uploads didn't record. + slog.WarnContext(ctx, "api.stream_unreadable", "file_id", fileID, "offset", rangeSpec.Offset, "length", rangeSpec.Length, "error", err) + return mapStreamError(err) + } return mapServiceError(err) } defer download.Reader.Close() + // Read the first bytes before committing the status line: when Telegram can't serve the + // part, the client gets a 502 instead of a 206 that ends early with nothing in it. + first := make([]byte, min(rangeSpec.Length, streamFirstRead)) + firstN := 0 + if len(first) > 0 { + firstN, err = io.ReadAtLeast(download.Reader, first, 1) + if firstN == 0 { + if isExpectedStreamEnd(ctx, err) { + return ctx.Err() + } + slog.WarnContext(ctx, "api.stream_unreadable", "file_id", fileID, "offset", rangeSpec.Offset, "length", rangeSpec.Length, "error", err) + return mapStreamError(err) + } + } + status := http.StatusOK if partial { status = http.StatusPartialContent @@ -160,7 +182,10 @@ func (h *RawHandler) streamFile(ctx context.Context, w http.ResponseWriter, user w.Header().Set("Last-Modified", file.ModTime.Time.UTC().Format(http.TimeFormat)) w.Header().Set("X-Accel-Buffering", "no") w.WriteHeader(status) - _, err = io.CopyN(w, download.Reader, rangeSpec.Length) + _, err = w.Write(first[:firstN]) + if err == nil { + _, err = io.CopyN(w, download.Reader, rangeSpec.Length-int64(firstN)) + } if err != nil && !isExpectedStreamEnd(ctx, err) { // Headers are already committed, so a JSON error response cannot replace // this stream. The Content-Length mismatch tells the client it was truncated. @@ -169,6 +194,35 @@ func (h *RawHandler) streamFile(ctx context.Context, w http.ResponseWriter, user return nil } +// streamFirstRead is how much of a stream is read before the response status is sent. +const streamFirstRead = 64 << 10 + +// mapStreamError is the response for a stream that failed before its first byte: errors with +// their own status keep it, anything else is Telegram (the storage behind the request) failing. +// The details, which name Telegram messages, stay in the log. +func mapStreamError(err error) error { + if isTelegramPartGone(err) { + return problem(http.StatusBadGateway, "telegram_part_unavailable", "Telegram could not serve a part of this file", err) + } + mapped := mapServiceError(err) + var p *Problem + if errors.As(mapped, &p) && p.Status == http.StatusInternalServerError { + return problem(http.StatusBadGateway, "telegram_read_failed", "reading this file from Telegram failed", err) + } + return mapped +} + +// isTelegramPartGone: Telegram answered, but can't serve the part: its message or document is +// gone, or -503 "Timeout" (the file DC couldn't read it), also when that was still being retried +// as the deadline ran out. +func isTelegramPartGone(err error) bool { + if errors.Is(err, telegramstore.ErrMessageNotFound) || errors.Is(err, telegramstore.ErrDocumentNotFound) { + return true + } + var rpc *tgerr.Error + return errors.As(err, &rpc) && rpc.Code == -503 +} + func isExpectedStreamEnd(ctx context.Context, err error) bool { return ctx.Err() != nil || isClientDisconnect(err) } diff --git a/internal/api/raw_stream_test.go b/internal/api/raw_stream_test.go new file mode 100644 index 00000000..d95cfb76 --- /dev/null +++ b/internal/api/raw_stream_test.go @@ -0,0 +1,153 @@ +package api + +import ( + "bytes" + "context" + "errors" + "fmt" + "io" + "net/http" + "net/http/httptest" + "testing" + + "github.com/google/uuid" + "github.com/gotd/td/tgerr" + "github.com/jackc/pgx/v5/pgtype" + + "github.com/tgdrive/teldrive/v2/internal/api/gen" + "github.com/tgdrive/teldrive/v2/internal/db/sqlcgen" + "github.com/tgdrive/teldrive/v2/internal/telegramstore" + "github.com/tgdrive/teldrive/v2/internal/transfer" +) + +func TestStreamFileAnswers502WhenTelegramFailsBeforeTheFirstByte(t *testing.T) { + for _, tc := range []struct { + name string + err error + code string + }{ + {"-503 behind the retry deadline", fmt.Errorf("download Telegram document chunk at 0: %w", errors.Join(context.DeadlineExceeded, tgerr.New(-503, "Timeout"))), "telegram_part_unavailable"}, + {"message gone", telegramstore.ErrMessageNotFound, "telegram_part_unavailable"}, + {"connection trouble", errors.New("engine was closed"), "telegram_read_failed"}, + } { + file, parts := streamTestFile(64 * 1024) + handler := &RawHandler{handler: &Handler{Downloader: transfer.NewDownloader(streamTestCatalog{file, parts}, &streamTestStorage{err: tc.err}, nil)}} + fileID, _ := uuid.FromBytes(file.ID.Bytes[:]) + recorder := httptest.NewRecorder() + err := handler.streamFile(context.Background(), recorder, file.UserID, fileID, file, gen.NewOptString("bytes=0-"), gen.OptETag{}, false) + var p *Problem + if !errors.As(err, &p) || p.Status != http.StatusBadGateway || p.Code != tc.code { + t.Errorf("%s: streamFile() = %v, want 502 %s", tc.name, err, tc.code) + continue + } + if len(recorder.Header().Get("Content-Range")) != 0 || recorder.Body.Len() != 0 { + t.Errorf("%s: the 206 was committed before the error", tc.name) + } + } +} + +func TestStreamFileAnswers502WhenAPartWithoutSizeIsGone(t *testing.T) { + file, parts := streamTestFile(64 * 1024) + parts[0].PlainSize, parts[0].StoredSize = pgtype.Int8{}, pgtype.Int8{} // an older upload: sizes resolved on open + handler := &RawHandler{handler: &Handler{Downloader: transfer.NewDownloader(streamTestCatalog{file, parts}, &streamTestStorage{metadataErr: telegramstore.ErrMessageNotFound}, nil)}} + fileID, _ := uuid.FromBytes(file.ID.Bytes[:]) + err := handler.streamFile(context.Background(), httptest.NewRecorder(), file.UserID, fileID, file, gen.NewOptString("bytes=0-"), gen.OptETag{}, false) + var p *Problem + if !errors.As(err, &p) || p.Status != http.StatusBadGateway || p.Code != "telegram_part_unavailable" { + t.Fatalf("streamFile() = %v, want 502 telegram_part_unavailable", err) + } +} + +func TestStreamFileStreamsWholeRangeAfterTheFirstRead(t *testing.T) { + file, parts := streamTestFile(300 * 1024) + payload := bytes.Repeat([]byte("abcdef"), 50*1024) + handler := &RawHandler{handler: &Handler{Downloader: transfer.NewDownloader(streamTestCatalog{file, parts}, &streamTestStorage{data: payload}, nil)}} + fileID, _ := uuid.FromBytes(file.ID.Bytes[:]) + recorder := httptest.NewRecorder() + if err := handler.streamFile(context.Background(), recorder, file.UserID, fileID, file, gen.NewOptString("bytes=10-"), gen.OptETag{}, false); err != nil { + t.Fatalf("streamFile() = %v", err) + } + if recorder.Code != http.StatusPartialContent || !bytes.Equal(recorder.Body.Bytes(), payload[10:]) { + t.Fatalf("code %d, %d bytes, want 206 with %d bytes", recorder.Code, recorder.Body.Len(), len(payload)-10) + } +} + +func TestMapStreamErrorKeepsOwnStatuses(t *testing.T) { + var p *Problem + if err := mapStreamError(context.DeadlineExceeded); !errors.As(err, &p) || p.Status != http.StatusGatewayTimeout { + t.Fatalf("deadline: %v, want 504", err) + } + if err := mapStreamError(io.ErrUnexpectedEOF); !errors.As(err, &p) || p.Status != http.StatusBadGateway { + t.Fatalf("short read: %v, want 502", err) + } +} + +func streamTestFile(size int64) (*sqlcgen.File, []*sqlcgen.FilePart) { + file := &sqlcgen.File{ + ID: pgtype.UUID{Bytes: uuid.New(), Valid: true}, UserID: 7, Name: "film.mkv", + Kind: sqlcgen.FileKindFile, Size: pgtype.Int8{Int64: size, Valid: true}, Status: sqlcgen.FileStatusActive, + } + return file, []*sqlcgen.FilePart{{ + PartNo: 1, ChannelID: 11, MessageID: 101, + PlainSize: pgtype.Int8{Int64: size, Valid: true}, StoredSize: pgtype.Int8{Int64: size, Valid: true}, + }} +} + +type streamTestCatalog struct { + file *sqlcgen.File + parts []*sqlcgen.FilePart +} + +func (c streamTestCatalog) Get(context.Context, int64, uuid.UUID) (*sqlcgen.File, error) { + return c.file, nil +} + +func (c streamTestCatalog) Parts(context.Context, int64, uuid.UUID) ([]*sqlcgen.FilePart, error) { + return c.parts, nil +} + +type streamTestStorage struct { + data []byte + err error + metadataErr error + metadataCalls int +} + +func (s *streamTestStorage) OpenRange(_ context.Context, request telegramstore.RangeRequest) (io.ReadCloser, error) { + if s.err != nil { + return io.NopCloser(failingReader{s.err}), nil + } + end := int64(len(s.data)) + if request.Length >= 0 && request.Offset+request.Length < end { + end = request.Offset + request.Length + } + return io.NopCloser(bytes.NewReader(s.data[request.Offset:end])), nil +} + +func (s *streamTestStorage) Upload(context.Context, telegramstore.UploadRequest) (telegramstore.StoredPart, error) { + return telegramstore.StoredPart{}, nil +} + +func (s *streamTestStorage) DeleteMessages(context.Context, int64, int64, []int64) error { return nil } + +func (s *streamTestStorage) CopyPart(context.Context, int64, int64, int64, int64) (telegramstore.StoredPart, error) { + return telegramstore.StoredPart{}, nil +} + +func (s *streamTestStorage) CreateChannel(context.Context, int64, string) (telegramstore.Channel, error) { + return telegramstore.Channel{}, nil +} + +func (s *streamTestStorage) DeleteChannel(context.Context, int64, int64) error { return nil } + +type failingReader struct{ err error } + +func (r failingReader) Read([]byte) (int, error) { return 0, r.err } + +func (s *streamTestStorage) Metadata(_ context.Context, request telegramstore.MetadataRequest) (telegramstore.StoredPart, error) { + s.metadataCalls++ + if s.metadataErr != nil { + return telegramstore.StoredPart{}, s.metadataErr + } + return telegramstore.StoredPart{ChannelID: request.ChannelID, MessageID: request.MessageID, Size: int64(len(s.data))}, nil +} diff --git a/internal/telegramstore/retry.go b/internal/telegramstore/retry.go index a446bf02..5abbcf8b 100644 --- a/internal/telegramstore/retry.go +++ b/internal/telegramstore/retry.go @@ -2,6 +2,7 @@ package telegramstore import ( "context" + "errors" "fmt" "log/slog" "strings" @@ -35,20 +36,22 @@ type retryMiddleware struct{ max int } func (m retryMiddleware) Handle(next tg.Invoker) telegram.InvokeFunc { return func(ctx context.Context, input bin.Encoder, output bin.Decoder) error { + var last error // the transient Telegram error being retried for attempt := 0; ; attempt++ { if err := ctx.Err(); err != nil { - return err + return deadlineWithCause(err, last) } err := next.Invoke(ctx, input, output) if err == nil { return nil } if ctx.Err() != nil { - return ctx.Err() + return deadlineWithCause(ctx.Err(), last) } if !isTransientTelegramError(err) { return err } + last = err clientID, _ := ClientID(ctx) if attempt >= m.max { slog.WarnContext(ctx, "telegram.rpc_retry_exhausted", @@ -66,13 +69,33 @@ func (m retryMiddleware) Handle(next tg.Invoker) telegram.InvokeFunc { select { case <-ctx.Done(): timer.Stop() - return ctx.Err() + return deadlineWithCause(ctx.Err(), err) case <-timer.C: } } } } +// deadlineWithCause keeps the Telegram error that was being retried when the caller's deadline +// ran out, so "Timeout" (-503) and friends aren't reported as a bare context deadline. A cancel +// stays a plain cancel: nobody is waiting for the reason. +func deadlineWithCause(ctxErr, last error) error { + if errors.Is(ctxErr, context.DeadlineExceeded) && last != nil { + return &retryDeadlineError{last: last, deadline: ctxErr} + } + return ctxErr +} + +// retryDeadlineError is a deadline that ran out while a Telegram error was being retried: it +// reads as that error and matches both (errors.Is context.DeadlineExceeded, tgerr.As). +type retryDeadlineError struct{ last, deadline error } + +func (e *retryDeadlineError) Error() string { + return e.last.Error() + " (retried until the deadline)" +} + +func (e *retryDeadlineError) Unwrap() []error { return []error{e.deadline, e.last} } + func telegramRetryDelay(attempt int) time.Duration { delay := 500 * time.Millisecond const maxDelay = 10 * time.Second diff --git a/internal/telegramstore/retry_test.go b/internal/telegramstore/retry_test.go index d11b7462..4bbdc808 100644 --- a/internal/telegramstore/retry_test.go +++ b/internal/telegramstore/retry_test.go @@ -160,3 +160,42 @@ func TestRetryMiddlewareRetriesConnectionResetByPeer(t *testing.T) { t.Fatalf("calls = %d, want 2", got) } } + +func TestRetryMiddlewareKeepsTelegramErrorWhenDeadlineEnds(t *testing.T) { + synctest.Test(t, func(t *testing.T) { + refusal := tgerr.New(-503, "Timeout") + invoke := retryMiddleware{max: 10}.Handle(telegram.InvokeFunc(func(context.Context, bin.Encoder, bin.Decoder) error { + return refusal + })) + ctx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + err := invoke(ctx, &tg.UploadGetFileRequest{}, &tg.UploadFileBox{}) + if !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("error = %v, want the deadline", err) + } + if rpc, ok := tgerr.As(err); !ok || rpc.Code != -503 { + t.Fatalf("error = %v, want Telegram's -503 kept with it", err) + } + + // A cancel stays a plain cancel. + ctx, cancel = context.WithCancel(context.Background()) + time.AfterFunc(time.Second, cancel) + err = invoke(ctx, &tg.UploadGetFileRequest{}, &tg.UploadFileBox{}) + if !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v, want canceled", err) + } + if _, ok := tgerr.As(err); ok { + t.Fatalf("a cancel carried the Telegram error: %v", err) + } + }) +} + +func TestRetryDeadlineErrorReadsAsTheTelegramError(t *testing.T) { + err := deadlineWithCause(context.DeadlineExceeded, tgerr.New(-503, "Timeout")) + if got, want := err.Error(), "rpc error code -503: Timeout (retried until the deadline)"; got != want { + t.Fatalf("Error() = %q, want %q", got, want) + } + if got := deadlineWithCause(context.Canceled, tgerr.New(-503, "Timeout")); got != context.Canceled { + t.Fatalf("cancel = %v, want the plain cancel", got) + } +} diff --git a/internal/transfer/download.go b/internal/transfer/download.go index dd4249dc..63b161dc 100644 --- a/internal/transfer/download.go +++ b/internal/transfer/download.go @@ -170,7 +170,7 @@ func (d *Downloader) resolveMissingPartSizes(ctx context.Context, session telegr UserID: userID, ChannelID: part.ChannelID, MessageID: part.MessageID, }) if err != nil { - return fmt.Errorf("resolve Telegram part %d metadata: %w", part.PartNo, err) + return fmt.Errorf("resolve Telegram part %d (message %d) metadata: %w", part.PartNo, part.MessageID, err) } plainSize := stored.Size if file.Encryption { @@ -428,7 +428,7 @@ func (r *downloadReader) openPartReader(segment downloadSegment, partOffset, spa }) } if err != nil { - return nil, fmt.Errorf("open Telegram part %d: %w", segment.part.PartNo, err) + return nil, fmt.Errorf("open Telegram part %d (message %d): %w", segment.part.PartNo, segment.part.MessageID, err) } return reader, nil }