Conversation
…t indefinite growth
fix(stream): suppress logging of client disconnect errors during stream copy (#189)
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughReplaces the legacy Telegram reader with a concurrent block-prefetch StreamPipe, introduces stream configuration and errors, adds timing/logging utilities, updates startup logging and CLI flags, bumps Go/tooling and many dependencies, and adjusts CI/release/goreleaser settings. Changes
Sequence Diagram(s)sequenceDiagram
participant Client as Client
participant Handler as HTTP/Route Handler
participant StreamPipe as StreamPipe
participant Workers as Prefetch Workers
participant Telegram as Telegram API
Client->>Handler: request file (start/end)
Handler->>StreamPipe: NewStreamPipe(ctx, client, location, start, end, log)
StreamPipe->>Workers: start background prefetch (spawn workers)
Workers->>Telegram: UploadGetFile (block N) [with retries/timeouts]
Telegram-->>Workers: block N bytes
Workers-->>StreamPipe: enqueue block N
Handler->>StreamPipe: Read(p)
StreamPipe->>StreamPipe: dequeue & trim block -> return bytes
Client->>Handler: connection close
Handler->>StreamPipe: Close()
StreamPipe->>Workers: cancel context -> stop prefetch
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 minutes Poem
🚥 Pre-merge checks | ✅ 2 | ❌ 2❌ Failed checks (1 warning, 1 inconclusive)
✅ Passed checks (2 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 8
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
.github/workflows/release.yml (1)
29-29:⚠️ Potential issue | 🟠 MajorGo version in workflow is outdated.
The workflow uses
go-version: 1.21but the project now targets Go 1.24+ (pergo.mod). This mismatch could cause build failures or miss language features used in the codebase.Proposed fix
- go-version: 1.21 + go-version: "1.24"
🤖 Fix all issues with AI agents
In @.github/workflows/release.yml:
- Line 41: The args line for the GitHub Actions step currently contains a stray
trailing double-quote which will pass an extra quote to goreleaser; in the step
that sets args: release ${{ github.event.inputs.goReleaserArgs }}" remove the
unmatched trailing quote so the value becomes properly closed (e.g., args:
release ${{ github.event.inputs.goReleaserArgs }}) and ensure no other
surrounding quotes are left unmatched in that step.
In @.goreleaser.yaml:
- Around line 26-51: The dockers_v2.tags list unconditionally includes the
"latest" tag which will be applied to prerelease builds; update the tags under
dockers_v2 to emit "latest" only for stable releases by replacing the
unconditional "latest" entry with a template that checks the .Prerelease field
and only adds "latest" when .Prerelease is empty (i.e., for stable releases);
modify the tags block (references: dockers_v2, tags, .Prerelease, .Tag/.Version)
accordingly so prerelease builds do not receive the "latest" tag.
In `@Dockerfile`:
- Line 1: Update the Dockerfile to match the Go toolchain version in go.mod by
changing the builder base image FROM golang:1.25-alpine3.21 to a Go 1.24 Alpine
image (use a tag that corresponds to go 1.24.x), and in the final stage replace
the root scratch runtime with a non-root numeric UID by creating a dedicated
unprivileged user (e.g., adduser or similar in the builder stage, copy
/etc/passwd or chown files as needed) and set USER <uid> in the final scratch
stage so the container does not run as root; ensure all files copied into the
final image have the correct ownership for that UID.
In `@go.mod`:
- Line 3: The GitHub Actions workflow uses Go 1.21 while go.mod declares go
1.24.0, causing a mismatch; update the release workflow's Go version by changing
the go-version key in .github/workflows/release.yml (the go-version: 1.21 entry)
to go-version: 1.24 (or 1.24.0) so the CI Go version matches the go.mod
requirement.
In `@internal/commands/start.go`:
- Line 29: Update the user-facing string passed to ext.ReplyTextString in the
start command to fix the typo "streamble" → "streamable": locate the call
ctx.Reply(..., ext.ReplyTextString("Hi, send me any file to get a direct
streamble link to that file."), ...) in internal/commands/start.go (the start
handler) and replace the message text so it reads "Hi, send me any file to get a
direct streamable link to that file." ensuring no other logic is changed.
In `@internal/commands/stream.go`:
- Around line 67-68: The code currently uses unchecked type assertions for
update.Updates[0] and update.Updates[1] when building messageID and doc, which
can panic; update the handler to first verify len(update.Updates) >= 2, then
perform type assertions with the comma-ok form for tg.UpdateMessageID and
tg.UpdateNewChannelMessage (and for casting Message to *tg.Message), check the
ok booleans and that the resulting pointers/Media are non-nil, and handle
failure paths gracefully (log the unexpected shape of update and return or skip
processing) so messageID and doc are only used when their types are confirmed.
In `@internal/routes/stream.go`:
- Around line 127-132: The response status is already committed before calling
stream.NewStreamPipe, so on NewStreamPipe failure http.Error cannot change the
status; move the header writes (calls to w.WriteHeader(http.StatusOK) and
w.WriteHeader(http.StatusPartialContent)) to after successfully creating the
pipe (i.e., after NewStreamPipe returns without error), or if you cannot
restructure flow now, remove the http.Error call and instead abort the request
by closing the underlying connection on pipe creation failure (use the
ResponseWriter's http.Hijacker to get the net.Conn and close it) and keep the
existing zap.Error log in NewStreamPipe error handling.
In `@internal/stream/pipe.go`:
- Around line 62-92: Validate the start/end bounds in NewStreamPipe: check that
start <= end at the top of the function (before computing totalBytes or calling
calculateBlockSize) and return a descriptive error (e.g., via fmt.Errorf or
errors.New) when start > end so you don't create negative totalBytes or pass a
negative size to calculateBlockSize; keep the rest of NewStreamPipe (ctx/cancel,
StreamPipe initialization, prefetch goroutine) unchanged.
🧹 Nitpick comments (3)
internal/utils/helpers.go (1)
32-41: Consider using typed error checks instead of string matching.String-based error detection is fragile across platforms and Go versions. Go's
syscallanderrorspackages allow more robust checks:Suggested approach
+import ( + "errors" + "syscall" +) + func IsClientDisconnectError(err error) bool { if err == nil { return false } - errStr := err.Error() - return strings.Contains(errStr, "connection was aborted") || - strings.Contains(errStr, "connection reset by peer") || - strings.Contains(errStr, "broken pipe") || - strings.Contains(errStr, "forcibly closed") + return errors.Is(err, syscall.ECONNRESET) || + errors.Is(err, syscall.EPIPE) || + errors.Is(err, syscall.ECONNABORTED) }Note: "forcibly closed" is a Windows-specific message (
WSAECONNRESET), whichsyscall.ECONNRESETshould cover on Windows as well. You may want to verify coverage for your target platforms.internal/stream/pipe.go (2)
136-143:rightTrimis computed relative to block alignment, which is correct — but consider documenting the invariant.The trimming math across first/last/single-block cases (lines 140–202) is correct after careful analysis, but the variable names
leftTrimandrightTrimare a bit misleading —leftTrimis an offset (bytes to skip), whilerightTrimis an upper-bound index (bytes to keep). A brief comment clarifying these semantics would help future maintainers.
238-281: Add jitter to exponential backoff to avoid thundering-herd retries.Multiple concurrent goroutines hitting the same retry cadence after a transient failure will all retry in lockstep, amplifying load on the Telegram API. Adding random jitter is a standard best practice for backoff.
Proposed fix
+import "math/rand/v2" + func (p *StreamPipe) downloadBlockWithRetry(offset int64) ([]byte, error) { var lastErr error backoff := 100 * time.Millisecond const maxBackoff = 15 * time.Second for attempt := 0; attempt < config.ValueOf.StreamMaxRetries; attempt++ { if p.ctx.Err() != nil { return nil, p.ctx.Err() } ctx, cancel := context.WithTimeout(p.ctx, time.Duration(config.ValueOf.StreamTimeoutSec)*time.Second) data, err := p.downloadBlock(ctx, offset) cancel() if err == nil { return data, nil } lastErr = err if p.ctx.Err() != nil { return nil, p.ctx.Err() } + // add jitter: 50-150% of backoff + jittered := time.Duration(float64(backoff) * (0.5 + rand.Float64())) select { - case <-time.After(backoff): + case <-time.After(jittered): backoff *= 2 if backoff > maxBackoff { backoff = maxBackoff } case <-p.ctx.Done(): return nil, p.ctx.Err() } }
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
.github/workflows/release.yml (1)
27-29:⚠️ Potential issue | 🟠 MajorAlign Go version in release workflow to 1.25 and update GitHub Actions to latest versions.
The release workflow builds with Go 1.21 while the Dockerfile uses
golang:1.25-alpine3.21, creating a mismatch. Binaries compiled with different Go versions can have subtle behavioural differences. Additionally,actions/checkout@v3andactions/setup-go@v4are behind their latest majors—v6andv6respectively.Proposed fix
- name: Set up Go - uses: actions/checkout@v3 + uses: actions/checkout@v6 - name: Set up Go - uses: actions/setup-go@v4 + uses: actions/setup-go@v6 with: - go-version: 1.21 + go-version: "1.25"
…r non-prerelease versions
…in sendLink function
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Fix all issues with AI agents
In `@go.mod`:
- Line 3: Update goreleaser.Dockerfile to use the same Go version as go.mod and
the main Dockerfile: replace the base image tag golang:1.21 with
golang:1.25-alpine3.21 so the build image matches the module's go 1.25.0
requirement and the production Dockerfile; ensure any related apk or dependency
commands remain compatible with the alpine3.21 variant.
🧹 Nitpick comments (4)
.github/workflows/release.yml (1)
22-27: Consider bumpingactions/checkoutandactions/setup-goto latest major versions.
actions/checkout@v3andactions/setup-go@v4have newer major versions available (v4andv5respectively) with performance and security improvements.internal/utils/timing.go (1)
9-17: Consider logging at Debug level instead of Info for high-frequency calls.This utility wraps
downloadBlockinpipe.go, which is called per block. For a large file stream, this could produce hundreds or thousands of Info-level log entries. Usinglog.Debug(...)would keep them available when needed without flooding production logs.Proposed change
defer func() { duration := time.Since(start).Milliseconds() - log.Info(funcName+" executed", zap.Int64("duration_ms", duration)) + log.Debug(funcName+" executed", zap.Int64("duration_ms", duration)) }()internal/stream/pipe.go (2)
186-203: Outer variablesleftTrim/rightTrimare mutated inside goroutine closures — fragile pattern.These variables are declared in
prefetch's scope (lines 142–143) and captured by reference in the concurrent goroutines. The mutations are currently safe because branch conditions ensure only one goroutine enters each case, but this is subtle and easily broken by future changes (e.g., adjusting batch composition).Consider passing them as goroutine-local copies, or computing the trim inline per block:
Sketch
go func(idx int) { defer wg.Done() blockNum := currentBlock + idx blockOffset := offset + int64(idx)*p.blockSize + localLeftTrim := leftTrim + localRightTrim := rightTrim data, err := p.downloadBlockWithRetry(blockOffset) ... // trim first/last block to exact range if totalBlocks == 1 { - if dataLen < rightTrim { - rightTrim = dataLen + if dataLen < localRightTrim { + localRightTrim = dataLen } - if leftTrim > dataLen { - leftTrim = dataLen + if localLeftTrim > dataLen { + localLeftTrim = dataLen } - data = data[leftTrim:rightTrim] + data = data[localLeftTrim:localRightTrim] } else if blockNum == 0 { ...
287-305: HandleUploadFileCDNRedirectresponse explicitly for better error clarity.The Telegram API's
UploadGetFilereturnstg.UploadFileClass, which has two implementations:*tg.UploadFileand*tg.UploadFileCDNRedirect. The current code only handles*tg.UploadFile, so CDN redirects fall through to thedefaultbranch and surface as a generic "unexpected response type" error. Adding an explicit case for*tg.UploadFileCDNRedirectwith a clearer error message improves debuggability.Suggested change
switch result := res.(type) { case *tg.UploadFile: return result.Bytes, nil + case *tg.UploadFileCDNRedirect: + return nil, fmt.Errorf("CDN redirect not supported (redirect to DC %d)", result.DCID) default: return nil, fmt.Errorf("unexpected response type: %T", res) }
There was a problem hiding this comment.
🧹 Nitpick comments (3)
goreleaser.Dockerfile (1)
1-2: Add a non-root USER directive (Trivy DS-0002).The container runs as root by default, which is flagged by static analysis. Since this image only runs the pre-built binary, add a non-root user for defense-in-depth.
Also, using the full
golangbase image (~800MB+) for a runtime-only container is quite heavy. Since goreleaser copies a pre-built binary, consider using a minimal base image likealpineorgcr.io/distroless/static-debian12instead.Proposed fix (alpine variant)
-FROM golang:1.25 +FROM alpine:3.21 +RUN addgroup -S appgroup && adduser -S appuser -G appgroup +USER appuser CMD ["/app/fsb"]Or with distroless:
-FROM golang:1.25 +FROM gcr.io/distroless/static-debian12:nonroot CMD ["/app/fsb"]internal/stream/pipe.go (1)
291-310:Limitfield type narrowing (int64→int) is safe here but worth a note.
p.blockSizeisint64andLimitisint. SinceblockSizemaxes out at 1 MiB this is fine on all platforms, but a brief comment (or an explicit cast with a bounds-check) would make the intent clearer for future maintainers.internal/routes/stream.go (1)
94-108: Consider deferringWriteHeaderuntil after successful pipe creation.Both Line 97 (
StatusOK) and Line 108 (StatusPartialContent) commit the response before the pipe is created on Line 130. IfNewStreamPipefails, the client sees a 200/206 with the promisedContent-Lengthbut receives no data — a silently broken download. MovingWriteHeader(and the content headers on Lines 118-127) to after Line 134 would let you return a proper 500 on failure.This is a structural improvement that can be deferred, but it would meaningfully improve error behavior for callers.
There was a problem hiding this comment.
🧹 Nitpick comments (3)
internal/stream/pipe.go (3)
108-122:ErrPipeDrainedswallows the actual download error.When
prefetchencounters a download failure, it logs the error and returns (closing the channel). The reader then receivesErrPipeDrainedwith no information about the root cause. Callers cannot distinguish a network timeout from an auth failure, etc.Consider propagating the fetch error through the pipe (e.g., store it in an
atomic.Valueor a guarded field and return it fromReadwhen the channel is closed).
170-211: Outer-scopeleftTrim/rightTrimmutated inside goroutines — safe today, fragile tomorrow.The trimming at lines 191–208 modifies
leftTrimandrightTrim(declared at lines 147–148) from within goroutines. Currently safe because each trim branch is entered by exactly one goroutine, but this invariant is implicit and easy to break in a future refactor. Moving the trim logic to afterwg.Wait()(operating onblocks[0]/blocks[batchSize-1]) would make the safety explicit and keep the goroutines pure data-fetchers.
292-313: CDN redirect "handling" is an error bail-out — consider documenting intent.The commit message advertises "handle CDN redirects," but the implementation returns an error. That's a valid choice (CDN file fetching requires additional crypto steps per the Telegram docs), but the mismatch between the commit message and behavior could confuse future contributors. A brief comment or
TODOnoting that full CDN support is deferred would clarify intent.
1 participant
Changes
Main Changes
Other Changes
Summary by CodeRabbit
New Features
Bug Fixes
Chores