Compare commits

...
14 Commits
Author SHA1 Message Date
Levi be0a395447 test: stabilize webhook synchronization and local cookie navigation (#1671) 2026-11-14 11:48:16 +08:00
Levi 506f689992 fix(extension): cancel stuck event handlers and unblock engine waiters (#1659) 2026-11-14 11:32:19 +08:00
Levi 38c2f1e8c2 refactor(ui): refine advanced options and settings feedback (#1661) 2026-11-14 11:30:33 +08:00
Levi 0520fd2baf fix(bt): upgrade torrent to v1.62.0 for WebTorrent announces (#1665) 2026-11-14 11:29:07 +08:00
Levi 6f7fd056ef fix(windows): release FFmpeg memory and capture native crashes (#1668) 2026-11-14 11:28:39 +08:00
Chandra ObulaReddy 6cd95507fd fix(http): decode HTML entities in filenames in one pass (#1651)
unescapeHTMLEntities applied its replacements while ranging over a map,
so the order changed from run to run. An escaped ampersand such as
"a<b.txt" came out as "a<b.txt" or, when & was replaced
first, as the double-decoded "a<b.txt".

Replace all entities in a single pass with strings.NewReplacer and
cover the escaped-ampersand cases in TestUnescapeHTMLEntities, running
each case repeatedly so an order-dependent implementation fails.
2026-11-12 09:23:13 +08:00
Lykinandxunguangzlj-cloud 89dfbe4c45 fix(download): separate torrent progress from transfer rates (#1654)
Co-authored-by: xunguangzlj-cloud <[email protected]>
2026-11-11 09:37:31 +08:00
Lykin 392ee36137 fix(web): wait for the saved analytics preference (#1656) 2026-11-10 15:00:15 +08:00
Konstantin OstrovskiiandLevi a34383a7e2 fix(http): wait for the prefetch goroutine to exit before reusing its file (#1606)
* fix(http): wait for the prefetch goroutine to exit before reusing its file

stopPrefetchAndCopyData waited at most 10 s for the Resolve prefetch, but a
stalled body read only times out after readTimeout (15 s). The caller then
closed and removed the temporary file while the prefetch goroutine could
still write to it.

Stopping the prefetch now closes the Resolve body, which interrupts a
pending read, and waits on a channel closed when the goroutine exits. Pause
uses the same path. The stop channel is closed under resolveRespLock so a
concurrent Pause and first-connection takeover cannot close it twice.

The proxy test helper now calls Setup before Resolve, as the downloader
does, so the Resolve request also goes through the proxy.

Fixes #1563

* refactor(http): wait on prefetchExitCh in completeFromResolveFallback

Replaces the 10 ms polling loop with the exit channel added in the
previous commit, so every place that waits for the prefetch uses it.

---------

Co-authored-by: Levi <[email protected]>
2026-11-09 11:14:49 +08:00
Silas MariuszandLevi 98879569f0 chore(deps): update modules with known vulnerabilities (#1612)
govulncheck (-tags nosqlite,web ./cmd/web/... and ./cmd/api/...) reports
40 reachable vulnerabilities in 10 modules on main. This update fixes 34
of them and keeps the go directive at 1.25.4, the version CI uses.

- github.com/go-git/go-git/v5 v5.8.1 -> v5.19.2: GO-2024-2456,
  GO-2024-2466, GO-2025-3367, GO-2025-3368, GO-2026-4473, GO-2026-4909,
  GO-2026-4910, GO-2026-5074, GO-2026-5105, GO-2026-5336, GO-2026-5496,
  GO-2026-5693, GO-2026-6213, GO-2026-6214
- github.com/go-git/go-billy/v5 v5.4.1 -> v5.9.2: GO-2026-5490,
  GO-2026-5597
- golang.org/x/crypto v0.48.0 -> v0.55.0: GO-2026-5013, GO-2026-5015,
  GO-2026-5017, GO-2026-5018, GO-2026-5019, GO-2026-5020, GO-2026-5021
- golang.org/x/net v0.50.0 -> v0.58.0: GO-2026-4559, GO-2026-5026,
  GO-2026-5942
- golang.org/x/text v0.34.0 -> v0.41.0: GO-2026-5970
- github.com/imroc/req/v3 v3.52.2 -> v3.58.0, which brings
  github.com/quic-go/quic-go v0.51.0 -> v0.59.0 (GO-2025-4017) and
  github.com/refraction-networking/utls v1.6.7 -> v1.8.2 (GO-2025-3638,
  GO-2026-4509, GO-2026-4512). req v3.52.2 does not build against a
  quic-go with the fix.
- github.com/pion/dtls/v3 v3.0.10 -> v3.1.4: GO-2026-6165
- github.com/pion/stun/v3 v3.1.1 -> v3.1.5: GO-2026-6163
- github.com/cloudflare/circl v1.6.1 -> v1.6.3: GO-2026-4550

The other module changes are what go mod tidy needs for these.

Still open, because the fixed releases need Go 1.26: golang.org/x/net
GO-2026-6603, GO-2026-6610, GO-2026-6611, GO-2026-6612 and GO-2026-6617
(fixed in v0.60.0), and golang.org/x/crypto GO-2026-6354 and GO-2026-6355
(fixed in v0.56.0). GO-2026-6610 affects every x/net before v0.60.0;
govulncheck does not report it as reachable with v0.50.0, but does with
v0.56.0 and v0.58.0.

Co-authored-by: Levi <[email protected]>
2026-11-09 10:52:49 +08:00
Levi 8ef3b63e8d fix(ui): respect mobile safe areas in tablet layouts (#1641) 2026-11-06 09:32:45 +08:00
Konstantin Ostrovskii d873cd411c fix(http): copy a fully prefetched body when Start runs before prefetchDone (#1608)
The prefetch goroutine sets prefetchDone only after one more read and
closing the body. A Start in that window took the whole file as the
fallback snapshot, expandConnections marked the task done, and the temp
file was never copied, leaving a zero-filled target.

Read prefetchSize once and treat the prefetch as complete when it reached
the known size.
2026-11-06 09:24:19 +08:00
Levi 9e9bcd4b01 fix(ci): keep beta Docker images off latest (#1645) 2026-11-04 14:41:53 +08:00
Silas Mariusz 5bd57608d2 fix(script): keep the task name in the script snapshot; wait for the full env output in tests (#1568) 2026-11-03 10:40:55 +08:00
82 changed files with 1785 additions and 460 deletions

No files matched your search

+2 -2
View File
@@ -13,7 +13,7 @@ on:
default: "false"
env:
GO_VERSION: "1.25.4"
GO_VERSION: "1.26.11"
FLUTTER_VERSION: "3.41.6"
GA4_MEASUREMENT_ID: ${{ secrets.GA4_MEASUREMENT_ID }}
GA4_API_SECRET: ${{ secrets.GA4_API_SECRET }}
@@ -798,5 +798,5 @@ jobs:
linux/arm64
linux/arm/v7
tags: |
liwei2633/gopeed:latest
${{ !contains(env.VERSION, 'beta') && 'liwei2633/gopeed:latest' || '' }}
liwei2633/gopeed:v${{ env.VERSION }}
+1 -1
View File
@@ -17,7 +17,7 @@ on:
- "Dockerfile"
env:
GO_VERSION: "1.25.4"
GO_VERSION: "1.26.11"
permissions:
contents: write
+4 -1
View File
@@ -28,7 +28,7 @@ on:
workflow_dispatch:
env:
GO_VERSION: "1.25.4"
GO_VERSION: "1.26.11"
FLUTTER_VERSION: "3.41.6"
jobs:
@@ -134,6 +134,9 @@ jobs:
if: runner.os == 'macOS'
with:
flutter-version: ${{ env.FLUTTER_VERSION }}
- name: Test Windows FFmpeg memory and stream lifecycle
if: runner.os == 'Windows'
run: go test -v -count=1 -timeout 180s ./pkg/download/engine/ffmpeg ./pkg/download/engine/inject/stream ./pkg/download/engine
- name: Install Linux WebView dependencies
if: runner.os == 'Linux'
run: |
+1 -1
View File
@@ -1,4 +1,4 @@
FROM golang:1.25.4-alpine3.22 AS go
FROM golang:1.26.11-alpine3.23 AS go
WORKDIR /app
COPY ./go.mod ./go.sum ./
RUN go mod download
+44 -47
View File
@@ -1,21 +1,22 @@
module github.com/GopeedLab/gopeed
go 1.25.4
go 1.26.0
require (
codeberg.org/gruf/go-ffmpreg v0.6.23-0.20260827095416-df1836cb5be0
github.com/GopeedLab/webview_go v0.0.0-20260918010912-0d4c27bb7914
github.com/anacrolix/torrent v1.60.1-0.20251217073903-486bcbe758e0
github.com/RoaringBitmap/roaring/v2 v2.16.0
github.com/anacrolix/torrent v1.62.0
github.com/armon/go-socks5 v0.0.0-20160902184237-e75332964ef5
github.com/bodgit/sevenzip v1.6.1
github.com/dop251/goja v0.0.0-20260106131823-651366fbe6e3
github.com/dop251/goja_nodejs v0.0.0-20240728170619-29b559befffc
github.com/fclairamb/ftpserverlib v0.31.0
github.com/go-git/go-git/v5 v5.8.1
github.com/go-git/go-git/v5 v5.19.2
github.com/google/jsonschema-go v0.4.3
github.com/gorilla/handlers v1.5.1
github.com/gorilla/mux v1.8.0
github.com/imroc/req/v3 v3.52.2
github.com/imroc/req/v3 v3.58.0
github.com/jlaffaye/ftp v0.2.4
github.com/matoous/go-nanoid/v2 v2.0.0
github.com/mattn/go-ieproxy v0.0.12
@@ -29,62 +30,61 @@ require (
github.com/tetratelabs/wazero v1.12.0
github.com/xiaoqidun/setft v0.0.0-20220310121541-be86327699ad
go.etcd.io/bbolt v1.4.3
golang.org/x/exp v0.0.0-20251219203646-944ab1f22d93
golang.org/x/exp v0.0.0-20260410095643-746e56fc9e2f
)
require (
github.com/cyphar/filepath-securejoin v0.6.1 // indirect
github.com/google/go-querystring v1.1.0 // indirect
github.com/onsi/gomega v1.36.3 // indirect
)
require (
github.com/STARRY-S/zip v0.2.3 // indirect
github.com/anacrolix/btree v0.1.1 // indirect
github.com/anacrolix/missinggo/v2 v2.10.0 // indirect
github.com/anacrolix/go-utp v0.0.0-20260908033909-9f1664acf866 // indirect
github.com/anacrolix/missinggo/v2 v2.11.0 // indirect
github.com/andybalholm/brotli v1.2.0 // indirect
github.com/bodgit/plumbing v1.3.0 // indirect
github.com/bodgit/windows v1.0.1 // indirect
github.com/cespare/xxhash/v2 v2.3.0 // indirect
github.com/coder/websocket v1.8.15 // indirect
github.com/dsnet/compress v0.0.2-0.20230904184137-39efe44ab707 // indirect
github.com/felixge/fgprof v0.9.5 // indirect
github.com/go-task/slim-sprig/v3 v3.0.0 // indirect
github.com/google/go-cmp v0.7.0 // indirect
github.com/hashicorp/errwrap v1.1.0 // indirect
github.com/hashicorp/go-multierror v1.1.1 // indirect
github.com/hashicorp/golang-lru/v2 v2.0.7 // indirect
github.com/icholy/digest v1.1.0 // indirect
github.com/klauspost/compress v1.18.0 // indirect
github.com/klauspost/compress v1.18.2 // indirect
github.com/klauspost/pgzip v1.2.6 // indirect
github.com/mikelolasagasti/xz v1.0.1 // indirect
github.com/minio/minlz v1.0.1 // indirect
github.com/nwaples/rardecode/v2 v2.2.0 // indirect
github.com/onsi/ginkgo/v2 v2.23.4 // indirect
github.com/pierrec/lz4/v4 v4.1.22 // indirect
github.com/pion/transport/v4 v4.0.1 // indirect
github.com/quic-go/qpack v0.5.1 // indirect
github.com/quic-go/quic-go v0.51.0 // indirect
github.com/refraction-networking/utls v1.6.7 // indirect
github.com/pion/transport/v4 v4.0.2 // indirect
github.com/quic-go/qpack v0.6.0 // indirect
github.com/quic-go/quic-go v0.59.0 // indirect
github.com/refraction-networking/utls v1.8.2 // indirect
github.com/segmentio/asm v1.1.3 // indirect
github.com/segmentio/encoding v0.5.4 // indirect
github.com/sorairolake/lzip-go v0.3.8 // indirect
github.com/ulikunitz/xz v0.5.15 // indirect
github.com/yosida95/uritemplate/v3 v3.0.2 // indirect
go.opentelemetry.io/auto/sdk v1.2.1 // indirect
go.uber.org/automaxprocs v1.6.0 // indirect
go.uber.org/mock v0.5.1 // indirect
go4.org v0.0.0-20230225012048-214862532bf5 // indirect
golang.org/x/mod v0.33.0
golang.org/x/oauth2 v0.35.0 // indirect
golang.org/x/sync v0.20.0 // indirect
golang.org/x/sync v0.23.0 // indirect
)
require (
dario.cat/mergo v1.0.0 // indirect
github.com/Microsoft/go-winio v0.6.1
github.com/ProtonMail/go-crypto v0.0.0-20230717121422-5aa5874ade95 // indirect
github.com/RoaringBitmap/roaring v1.9.4
github.com/acomagu/bufpipe v1.0.4 // indirect
github.com/Microsoft/go-winio v0.6.2
github.com/ProtonMail/go-crypto v1.1.6 // indirect
github.com/alecthomas/atomic v0.1.0-alpha2 // indirect
github.com/anacrolix/chansync v0.7.0 // indirect
github.com/anacrolix/dht/v2 v2.23.0 // indirect
github.com/anacrolix/chansync v0.8.0 // indirect
github.com/anacrolix/dht/v2 v2.23.1-0.20260525063928-ec3a9bd99456 // indirect
github.com/anacrolix/envpprof v1.5.0 // indirect
github.com/anacrolix/generics v0.2.0 // indirect
github.com/anacrolix/go-libutp v1.3.2 // indirect
github.com/anacrolix/generics v0.2.0
github.com/anacrolix/go-libutp v1.5.2-0.20260908013213-836dd42cdde6 // indirect
github.com/anacrolix/log v0.17.1-0.20251118025802-918f1157b7bb // indirect
github.com/anacrolix/missinggo v1.3.0 // indirect
github.com/anacrolix/missinggo/perf v1.0.0 // indirect
@@ -93,13 +93,12 @@ require (
github.com/anacrolix/stm v0.5.0 // indirect
github.com/anacrolix/sync v0.6.0 // indirect
github.com/anacrolix/upnp v0.1.4 // indirect
github.com/anacrolix/utp v0.2.0 // indirect
github.com/bahlo/generic-list-go v0.2.0 // indirect
github.com/benbjohnson/immutable v0.4.3 // indirect
github.com/bits-and-blooms/bitset v1.24.4 // indirect
github.com/bradfitz/iter v0.0.0-20191230175014-e8f45d346db8 // indirect
github.com/cespare/xxhash v1.1.0 // indirect
github.com/cloudflare/circl v1.6.1 // indirect
github.com/cloudflare/circl v1.6.3 // indirect
github.com/davecgh/go-spew v1.1.1 // indirect
github.com/dlclark/regexp2 v1.11.5 // indirect
github.com/dustin/go-humanize v1.0.1 // indirect
@@ -107,17 +106,16 @@ require (
github.com/emirpasic/gods v1.18.1 // indirect
github.com/felixge/httpsnoop v1.0.1 // indirect
github.com/go-git/gcfg v1.5.1-0.20230307220236-3a3c6141e376 // indirect
github.com/go-git/go-billy/v5 v5.4.1 // indirect
github.com/go-git/go-billy/v5 v5.9.2 // indirect
github.com/go-llsqlite/adapter v0.2.0 // indirect
github.com/go-llsqlite/crawshaw v0.6.0 // indirect
github.com/go-logr/logr v1.4.3 // indirect
github.com/go-logr/logr v1.4.4 // indirect
github.com/go-logr/stdr v1.2.2 // indirect
github.com/go-sourcemap/sourcemap v2.1.4+incompatible // indirect
github.com/golang/groupcache v0.0.0-20210331224755-41bb18bfe9da // indirect
github.com/golang/groupcache v0.0.0-20241129210726-2c02b8208cf8 // indirect
github.com/google/btree v1.1.3 // indirect
github.com/google/pprof v0.0.0-20260115054156-294ebfa9ad83 // indirect
github.com/google/uuid v1.6.0 // indirect
github.com/gorilla/websocket v1.5.3 // indirect
github.com/huandu/xstrings v1.5.0 // indirect
github.com/jbenet/go-context v0.0.0-20150711004518-d14ea06fba99 // indirect
github.com/kevinburke/ssh_config v1.2.0 // indirect
@@ -131,7 +129,7 @@ require (
github.com/multiformats/go-varint v0.1.0 // indirect
github.com/ncruces/go-strftime v1.0.0 // indirect
github.com/pion/datachannel v1.6.0 // indirect
github.com/pion/dtls/v3 v3.0.10 // indirect
github.com/pion/dtls/v3 v3.1.4 // indirect
github.com/pion/ice/v4 v4.2.0 // indirect
github.com/pion/interceptor v0.1.42 // indirect
github.com/pion/logging v0.2.4 // indirect
@@ -142,28 +140,27 @@ require (
github.com/pion/sctp v1.9.1 // indirect
github.com/pion/sdp/v3 v3.0.17 // indirect
github.com/pion/srtp/v3 v3.0.10 // indirect
github.com/pion/stun/v3 v3.1.1 // indirect
github.com/pion/stun/v3 v3.1.5 // indirect
github.com/pion/turn/v4 v4.1.4 // indirect
github.com/pion/webrtc/v4 v4.2.2-0.20260109001657-a5962f314db7 // indirect
github.com/pjbgf/sha1cd v0.3.0 // indirect
github.com/pjbgf/sha1cd v0.6.0 // indirect
github.com/pkg/browser v0.0.0-20240102092130-5ac0b6a4141c
github.com/protolambda/ctxlock v0.1.0 // indirect
github.com/remyoudompheng/bigfft v0.0.0-20230129092748-24d4a6f8daec // indirect
github.com/sergi/go-diff v1.1.0 // indirect
github.com/skeema/knownhosts v1.2.0 // indirect
github.com/sergi/go-diff v1.3.2-0.20230802210424-5b0b94c5c0d3 // indirect
github.com/skeema/knownhosts v1.3.1 // indirect
github.com/spaolacci/murmur3 v1.1.0 // indirect
github.com/tidwall/btree v1.8.1 // indirect
github.com/wlynxg/anet v0.0.5 // indirect
github.com/xanzy/ssh-agent v0.3.3 // indirect
go.opentelemetry.io/otel v1.39.0 // indirect
go.opentelemetry.io/otel/metric v1.39.0 // indirect
go.opentelemetry.io/otel/trace v1.39.0 // indirect
golang.org/x/crypto v0.48.0 // indirect
golang.org/x/net v0.50.0
golang.org/x/sys v0.44.0 // indirect
golang.org/x/text v0.34.0
golang.org/x/time v0.15.0 // indirect
golang.org/x/tools v0.42.0 // indirect
go.opentelemetry.io/otel v1.45.0 // indirect
go.opentelemetry.io/otel/metric v1.45.0 // indirect
go.opentelemetry.io/otel/trace v1.45.0 // indirect
golang.org/x/crypto v0.57.0 // indirect
golang.org/x/net v0.60.0
golang.org/x/sys v0.48.0 // indirect
golang.org/x/text v0.42.0
golang.org/x/time v0.15.0
gopkg.in/warnings.v0 v0.1.2 // indirect
lukechampine.com/blake3 v1.4.1 // indirect
modernc.org/libc v1.67.4 // indirect
+91 -163
View File
File diff suppressed because it is too large. Load diff
+4
View File
@@ -46,6 +46,10 @@ type Uploader interface {
WaitUpload() error
}
type DownloadCounter interface {
ReceivedBytes() int64
}
// FetcherMeta defines the meta information of a fetcher.
type FetcherMeta struct {
Req *base.Request `json:"req"`
+76
View File
@@ -290,6 +290,82 @@ func TestBTDownloadSelectedFiles(t *testing.T) {
if f.Meta().Res.Size != 48*1024 {
t.Fatal("selected resource size is incorrect")
}
if downloaded := f.Progress().TotalDownloaded(); downloaded != 48*1024 || f.ReceivedBytes() <= downloaded {
t.Fatalf("selected progress = %d, received traffic = %d; want 48 KiB of progress and traffic across the shared piece", downloaded, f.ReceivedBytes())
}
}
func TestBTProgressOnlyCountsVerifiedFileBytes(t *testing.T) {
const fileSize = 56 * 1024
fixture := makeBTFixtureWithPieceSize(t, map[string][]byte{
"a.bin": btPayload(fileSize, "verified-a"),
"b.bin": btPayload(fileSize, "verified-b"),
}, 16*1024, false, true, 64*1024)
f := btDownloadFetcher(t, nil)
btResolve(t, f, fixture.Magnet)
dir := t.TempDir()
f.Meta().Opts = &base.Options{Path: dir}
btStart(t, f)
deadline := time.Now().Add(8 * time.Second)
for f.torrent.Files()[0].BytesCompleted() == 0 && time.Now().Before(deadline) {
time.Sleep(10 * time.Millisecond)
}
btPause(t, f)
if f.torrent.Files()[0].BytesCompleted() == 0 || f.torrent.Piece(0).State().Complete {
t.Fatal("expected some received bytes in a piece that is still incomplete")
}
if f.ReceivedBytes() == 0 {
t.Fatal("received traffic must be visible before a piece is verified")
}
if progress := f.Progress(); len(progress) != 2 || progress[0] != 0 {
t.Errorf("progress before the first file's piece is verified = %v, want zero for the first file", progress)
}
fixture.Limiter.SetLimit(rate.Inf)
btStart(t, f)
btAwaitDone(t, f)
fixture.AssertFiles(t, dir)
progress := f.Progress()
if len(progress) != 2 || progress[0] != fileSize || progress[1] != fileSize {
t.Fatalf("verified progress at completion = %v, want %d for each file", progress, fileSize)
}
}
func TestBTUploadedBytesSamplesLiveTraffic(t *testing.T) {
payload := btPayload(48*1024, "upload-counter")
fixture := newBTFixture(t, map[string][]byte{"": payload}, 0)
const saved = 1024
f := &Fetcher{
torrent: fixture.Torrent,
data: &fetcherData{SeedBytes: saved},
seedBaseBytes: saved,
}
f.torrentReady.Store(true)
f.torrentUpload.Store(true)
cfg := torrent.TestingConfig(t)
cfg.DataDir = t.TempDir()
cfg.DisableTrackers = true
cfg.DisableUTP = true
cfg.DisableIPv6 = true
peer, err := torrent.NewClient(cfg)
if err != nil {
t.Fatal(err)
}
t.Cleanup(func() { peer.Close() })
download, _, err := peer.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(fixture.MetaInfo))
if err != nil {
t.Fatal(err)
}
download.AddPeers([]torrent.PeerInfo{{Addr: fixture.Seeder.ListenAddrs()[0]}})
download.DownloadAll()
select {
case <-download.Complete().On():
case <-time.After(10 * time.Second):
t.Fatal("loopback peer did not complete")
}
if got, want := f.UploadedBytes(), int64(saved+len(payload)); got != want {
t.Fatalf("uploaded bytes = %d, want %d including the saved bytes", got, want)
}
}
func TestBTPauseAndResume(t *testing.T) {
+64 -11
View File
@@ -10,6 +10,7 @@ import (
"os"
"path/filepath"
"runtime"
"sort"
"strings"
"sync"
"sync/atomic"
@@ -20,7 +21,7 @@ import (
"github.com/GopeedLab/gopeed/pkg/base"
"github.com/GopeedLab/gopeed/pkg/protocol/bt"
"github.com/GopeedLab/gopeed/pkg/util"
"github.com/RoaringBitmap/roaring"
"github.com/RoaringBitmap/roaring/v2"
"github.com/anacrolix/torrent"
"github.com/anacrolix/torrent/metainfo"
"github.com/anacrolix/torrent/storage"
@@ -42,9 +43,10 @@ type Fetcher struct {
ctl *controller.Controller
config *config
torrent *torrent.Torrent
meta *fetcher.FetcherMeta
data *fetcherData
torrent *torrent.Torrent
meta *fetcher.FetcherMeta
data *fetcherData
seedBaseBytes int64
torrentReady atomic.Bool
torrentUpload atomic.Bool
@@ -463,14 +465,46 @@ func (f *Fetcher) Progress() fetcher.Progress {
if !f.torrentReady.Load() {
return append(fetcher.Progress(nil), f.data.Progress...)
}
pieceSize := f.torrent.Info().PieceLength
var offset int64
var verified []verifiedByteRange
for _, run := range f.torrent.PieceStateRuns() {
end := offset + int64(run.Length)*pieceSize
if pieceVerified(run.PieceState) {
verified = append(verified, verifiedByteRange{start: offset, end: end})
}
offset = end
}
files := f.torrent.Files()
for i := range f.data.Progress {
selectIndex := f.meta.Opts.SelectFiles[i]
file := f.torrent.Files()[selectIndex]
f.data.Progress[i] = file.BytesCompleted()
file := files[selectIndex]
f.data.Progress[i] = verifiedFileBytes(verified, file.Offset(), file.Length())
}
return append(fetcher.Progress(nil), f.data.Progress...)
}
type verifiedByteRange struct {
start, end int64
}
func verifiedFileBytes(verified []verifiedByteRange, offset, length int64) int64 {
first := sort.Search(len(verified), func(i int) bool { return verified[i].end > offset })
fileEnd := offset + length
var completed int64
for _, run := range verified[first:] {
if run.start >= fileEnd {
break
}
completed += min(fileEnd, run.end) - max(offset, run.start)
}
return completed
}
func pieceVerified(state torrent.PieceState) bool {
return state.Ok && state.Complete && !state.Hashing && !state.QueuedForHash
}
func (f *Fetcher) Wait() error {
f.mu.Lock()
var closed <-chan struct{}
@@ -587,7 +621,7 @@ func (f *Fetcher) isDoneLocked() bool {
file := f.torrent.Files()[index]
for i := file.BeginPieceIndex(); i < file.EndPieceIndex(); i++ {
state := f.torrent.Piece(i).State()
if !state.Ok || !state.Complete || state.Hashing || state.QueuedForHash {
if !pieceVerified(state) {
return false
}
}
@@ -685,11 +719,13 @@ func (f *Fetcher) Upload() (err error) {
}
func (f *Fetcher) doUpload(fromUpload bool) {
f.mu.Lock()
if !f.torrentUpload.CompareAndSwap(false, true) {
f.mu.Unlock()
return
}
f.mu.Lock()
initialBytes, initialTime := f.data.SeedBytes, f.data.SeedTime
f.seedBaseBytes = f.data.SeedBytes
initialTime := f.data.SeedTime
torrentClosed := f.torrent.Closed()
f.mu.Unlock()
var doneTime int64
@@ -707,8 +743,7 @@ func (f *Fetcher) doUpload(fromUpload bool) {
f.mu.Unlock()
continue
}
stats := f.torrentStats()
f.data.SeedBytes = initialBytes + stats.BytesWrittenData.Int64()
f.refreshSeedBytesLocked()
if !fromUpload && !f.isDoneLocked() {
f.mu.Unlock()
continue
@@ -741,9 +776,27 @@ func (f *Fetcher) torrentStats() torrent.TorrentStats {
func (f *Fetcher) UploadedBytes() int64 {
f.mu.Lock()
defer f.mu.Unlock()
if f.torrentReady.Load() && f.torrentUpload.Load() {
f.refreshSeedBytesLocked()
}
return f.data.SeedBytes
}
func (f *Fetcher) refreshSeedBytesLocked() {
stats := f.torrentStats()
f.data.SeedBytes = max(f.data.SeedBytes, f.seedBaseBytes+stats.BytesWrittenData.Int64())
}
func (f *Fetcher) ReceivedBytes() int64 {
f.mu.Lock()
defer f.mu.Unlock()
if !f.torrentReady.Load() {
return 0
}
stats := f.torrentStats()
return stats.BytesReadUsefulData.Int64()
}
func (f *Fetcher) WaitUpload() (err error) {
<-f.uploadDoneCh
return nil
+6 -1
View File
@@ -46,6 +46,11 @@ func newBTHybridFixture(t testing.TB, payload []byte) *btFixture {
}
func makeBTFixture(t testing.TB, files map[string][]byte, bytesPerSecond int, v2, private bool) *btFixture {
t.Helper()
return makeBTFixtureWithPieceSize(t, files, bytesPerSecond, v2, private, 16*1024)
}
func makeBTFixtureWithPieceSize(t testing.TB, files map[string][]byte, bytesPerSecond int, v2, private bool, pieceSize int64) *btFixture {
t.Helper()
source := t.TempDir()
name := "payload.bin"
@@ -64,7 +69,7 @@ func makeBTFixture(t testing.TB, files map[string][]byte, bytesPerSecond int, v2
}
expected[filepath.Join(name, filepath.FromSlash(path))] = append([]byte(nil), data...)
}
info := metainfo.Info{PieceLength: 16 * 1024}
info := metainfo.Info{PieceLength: pieceSize}
if err := info.BuildFromFilePath(root); err != nil {
t.Fatal(err)
}
+1 -1
View File
@@ -6,7 +6,7 @@ import (
"github.com/GopeedLab/gopeed/internal/fetcher"
"github.com/GopeedLab/gopeed/pkg/base"
protocolbt "github.com/GopeedLab/gopeed/pkg/protocol/bt"
"github.com/RoaringBitmap/roaring"
"github.com/RoaringBitmap/roaring/v2"
"github.com/anacrolix/torrent"
"github.com/anacrolix/torrent/storage"
)
+45 -34
View File
@@ -314,6 +314,7 @@ type Fetcher struct {
prefetchDone atomic.Bool // Prefetch completed or stopped
prefetchErr error // Error during prefetch (if any)
prefetchStopCh chan struct{} // Signal to stop prefetch
prefetchExitCh chan struct{} // Closed when the prefetch goroutine exits
resolveFallback atomic.Bool // Resolve prefetch is still available until the first Range request takes over
// Target file
@@ -562,6 +563,7 @@ func (f *Fetcher) Resolve(req *base.Request) error {
// For non-range resources, the response will be used directly in Start
if res.Range && reuseResolveBody {
f.prefetchStopCh = make(chan struct{})
f.prefetchExitCh = make(chan struct{})
f.resolveFallback.Store(true)
go f.asyncPrefetch()
}
@@ -579,6 +581,9 @@ func (f *Fetcher) Resolve(req *base.Request) error {
func (f *Fetcher) asyncPrefetch() {
defer func() {
f.prefetchDone.Store(true)
if f.prefetchExitCh != nil {
close(f.prefetchExitCh)
}
}()
// Get the resolve response
@@ -654,23 +659,38 @@ func (f *Fetcher) asyncPrefetch() {
}
}
// stopPrefetch signals the async prefetch to stop and waits until it exits.
// Closing the resolve body unblocks a pending Read, so a stalled server cannot
// keep the prefetch goroutine writing to the temporary file after we reuse it.
func (f *Fetcher) stopPrefetch() {
if f.prefetchStopCh == nil {
return
}
// Pause and the first connection's takeover can stop concurrently, so the
// check-and-close must be under the lock to avoid closing the channel twice.
f.resolveRespLock.Lock()
select {
case <-f.prefetchStopCh:
// Already closed
default:
close(f.prefetchStopCh)
}
if f.resolveResp != nil {
f.resolveResp.Body.Close()
f.resolveResp = nil
}
f.resolveRespLock.Unlock()
if f.prefetchExitCh != nil {
<-f.prefetchExitCh
}
}
// stopPrefetchAndGetData stops the async prefetch and returns prefetched bytes
// It also copies prefetched data to the target file
func (f *Fetcher) stopPrefetchAndCopyData() int64 {
// Signal prefetch to stop (safely)
if f.prefetchStopCh != nil {
select {
case <-f.prefetchStopCh:
// Already closed
default:
close(f.prefetchStopCh)
}
}
// Wait for prefetch to finish (with timeout)
for i := 0; f.prefetchStopCh != nil && i < 1000 && !f.prefetchDone.Load(); i++ {
time.Sleep(10 * time.Millisecond)
}
f.stopPrefetch()
prefetched := f.prefetchSize.Load()
if prefetched == 0 {
@@ -872,12 +892,16 @@ func (f *Fetcher) doStart() error {
// For non-range resources, the response will be used directly
var prefetchedBytes int64
if f.meta.Res.Range {
resolveComplete := f.knownSize() > 0 && f.prefetchDone.Load() && f.prefetchErr == nil && f.prefetchSize.Load() >= f.knownSize()
// Read the size once and judge completeness by it alone: Resolve may have
// written every byte before it sets prefetchDone, and a snapshot that
// covers the whole file would otherwise finish without copying it.
prefetched := f.prefetchSize.Load()
resolveComplete := f.knownSize() > 0 && prefetched >= f.knownSize()
if f.resolveFallback.Load() && !resolveComplete {
// Keep Resolve downloading until the first Range response has been
// validated. The Range request starts at a stable snapshot; a valid
// takeover discards any overlap downloaded by Resolve in the meantime.
prefetchedBytes = f.prefetchSize.Load()
prefetchedBytes = prefetched
} else {
// Resolve already completed, or no live Resolve fallback remains.
prefetchedBytes = f.stopPrefetchAndCopyData()
@@ -1627,11 +1651,10 @@ func (f *Fetcher) completeFromResolveFallback(conn *connection) bool {
return false
}
for !f.prefetchDone.Load() {
if conn.ctx.Err() != nil {
return false
}
time.Sleep(10 * time.Millisecond)
select {
case <-f.prefetchExitCh:
case <-conn.ctx.Done():
return false
}
knownSize := f.hasKnownSize()
if f.prefetchErr != nil || (knownSize && f.prefetchSize.Load() != f.meta.Res.Size) {
@@ -2438,14 +2461,7 @@ func (f *Fetcher) Pause() error {
}
// Stop prefetch if running
if f.prefetchStopCh != nil {
select {
case <-f.prefetchStopCh:
// Already closed
default:
close(f.prefetchStopCh)
}
}
f.stopPrefetch()
// Wait for downloadLoop to exit first (it will call wg.Wait internally)
if f.downloadLoopDone != nil {
@@ -2455,11 +2471,6 @@ func (f *Fetcher) Pause() error {
// Wait for all connection goroutines to stop
f.wg.Wait()
// Wait for prefetch to finish
for f.prefetchStopCh != nil && !f.prefetchDone.Load() {
time.Sleep(10 * time.Millisecond)
}
// If Pause wins before the first Range response is validated, materialize
// the Resolve prefix before discarding its temporary file. The first
// connection was allocated after the earlier prefetch snapshot, so advance
+5 -4
View File
@@ -3261,16 +3261,17 @@ func downloadResume(listener net.Listener, connections int, t *testing.T) {
}
func downloadWithProxy(httpListener net.Listener, proxyListener net.Listener, t *testing.T) {
fetcher := downloadReady(httpListener, 4, t)
ctl := controller.NewController()
ctl.GetProxy = func(requestProxy *base.RequestProxy) func(*gohttp.Request) (*url.URL, error) {
// Setup must run before Resolve, as in the downloader, so the Resolve
// request also goes through the proxy.
f := buildFetcher()
f.ctl.GetProxy = func(requestProxy *base.RequestProxy) func(*gohttp.Request) (*url.URL, error) {
return (&base.DownloaderProxyConfig{
Enable: true,
Scheme: "socks5",
Host: proxyListener.Addr().String(),
}).ToHandler()
}
fetcher.Setup(ctl)
fetcher := doDownloadReady(f, httpListener, 4, t)
err := fetcher.Start()
if err != nil {
t.Fatal(err)
+16 -3
View File
@@ -125,13 +125,26 @@ func TestUnescapeHTMLEntities(t *testing.T) {
in: "normal.txt",
want: "normal.txt",
},
{
name: "escaped ampersand before entity name is decoded once",
in: "a&amp;lt;b.txt",
want: "a&lt;b.txt",
},
{
name: "escaped ampersand before quote entity is decoded once",
in: "a&amp;quot;b.txt",
want: "a&quot;b.txt",
},
}
for _, tt := range tests {
t.Run(tt.name, func(t *testing.T) {
got := unescapeHTMLEntities(tt.in)
if got != tt.want {
t.Errorf("unescapeHTMLEntities() = %q, want %q", got, tt.want)
// Run repeatedly so that an order-dependent implementation fails reliably.
for i := 0; i < 200; i++ {
got := unescapeHTMLEntities(tt.in)
if got != tt.want {
t.Fatalf("unescapeHTMLEntities() = %q, want %q (run %d)", got, tt.want, i)
}
}
})
}
+12 -15
View File
@@ -408,24 +408,21 @@ func decodeFilenameParam(filename string) string {
return decoded
}
// htmlEntityReplacer replaces every entity in a single left-to-right pass, so an
// escaped ampersand such as "&amp;lt;" becomes "&lt;" and is not decoded twice.
var htmlEntityReplacer = strings.NewReplacer(
"&amp;", "&",
"&lt;", "<",
"&gt;", ">",
"&quot;", "\"",
"&#39;", "'",
"&apos;", "'",
)
// unescapeHTMLEntities unescapes common HTML entities in filenames
// This handles cases where servers HTML-encode special characters like & to &amp;
func unescapeHTMLEntities(s string) string {
// Common HTML entities that might appear in filenames
replacements := map[string]string{
"&amp;": "&",
"&lt;": "<",
"&gt;": ">",
"&quot;": "\"",
"&#39;": "'",
"&apos;": "'",
}
result := s
for entity, char := range replacements {
result = strings.ReplaceAll(result, entity, char)
}
return result
return htmlEntityReplacer.Replace(s)
}
// tryDecodeGBK attempts to decode string as GBK encoding
@@ -0,0 +1,52 @@
package http
import (
"bytes"
gohttp "net/http"
"net/http/httptest"
"os"
"testing"
"time"
"github.com/GopeedLab/gopeed/pkg/base"
"github.com/GopeedLab/gopeed/pkg/protocol/http"
)
// Start can run after the Resolve prefetch wrote every byte but before it set
// prefetchDone. The prefetched body must still be copied to the target.
func TestFetcher_StartBeforePrefetchDoneCopiesFullBody(t *testing.T) {
payload := []byte("prefetched before Start")
server := httptest.NewServer(gohttp.HandlerFunc(func(w gohttp.ResponseWriter, r *gohttp.Request) {
gohttp.ServeContent(w, r, "f.txt", time.Time{}, bytes.NewReader(payload))
}))
defer server.Close()
f := buildFetcher()
f.Meta().Opts = &base.Options{Path: t.TempDir(), Name: "f.txt", Extra: &http.OptsExtra{Connections: 1}}
if err := f.Resolve(&base.Request{URL: server.URL + "/f.txt"}); err != nil {
t.Fatal(err)
}
for !f.prefetchDone.Load() {
time.Sleep(time.Millisecond)
}
// Reopen the window between the last write and prefetchDone.
f.prefetchDone.Store(false)
go func() {
time.Sleep(50 * time.Millisecond)
f.prefetchDone.Store(true)
}()
if err := f.Start(); err != nil {
t.Fatal(err)
}
if err := f.Wait(); err != nil {
t.Fatal(err)
}
got, err := os.ReadFile(f.meta.SingleFilepath())
if err != nil {
t.Fatal(err)
}
if !bytes.Equal(got, payload) {
t.Fatalf("content = %q, want %q", got, payload)
}
}
@@ -0,0 +1,56 @@
package http
import (
"bytes"
"fmt"
gohttp "net/http"
"net/http/httptest"
"testing"
"time"
"github.com/GopeedLab/gopeed/pkg/base"
"github.com/GopeedLab/gopeed/pkg/protocol/http"
)
// A stalled Resolve body must not keep the prefetch goroutine alive after
// stopPrefetchAndCopyData returns, or it writes to a file we already removed.
func TestFetcher_StopPrefetchUnblocksStalledRead(t *testing.T) {
prefix := bytes.Repeat([]byte("p"), 64*1024)
stall := make(chan struct{})
defer close(stall)
server := httptest.NewServer(gohttp.HandlerFunc(func(w gohttp.ResponseWriter, r *gohttp.Request) {
w.Header().Set(base.HttpHeaderAcceptRanges, base.HttpHeaderBytes)
w.Header().Set(base.HttpHeaderContentLength, fmt.Sprintf("%d", len(prefix)*4))
w.WriteHeader(gohttp.StatusOK)
_, _ = w.Write(prefix)
w.(gohttp.Flusher).Flush()
select {
case <-stall:
case <-r.Context().Done():
}
}))
defer server.Close()
f := buildFetcher()
f.Meta().Opts = &base.Options{
Path: t.TempDir(),
Name: "stalled.data",
Extra: &http.OptsExtra{Connections: 1},
}
if err := f.Resolve(&base.Request{URL: server.URL + "/stalled.data"}); err != nil {
t.Fatal(err)
}
for f.prefetchSize.Load() < int64(len(prefix)) {
time.Sleep(time.Millisecond)
}
start := time.Now()
f.stopPrefetch()
if elapsed := time.Since(start); elapsed > 2*time.Second {
t.Fatalf("stopPrefetch took %v, want it to interrupt the stalled read", elapsed)
}
if !f.prefetchDone.Load() {
t.Fatal("prefetch goroutine still running after stopPrefetch")
}
}
+5 -8
View File
@@ -25,7 +25,6 @@ const defaultAvailabilityTimeout = time.Minute
type ContractOptions struct {
AvailabilityTimeout time.Duration
CookieDomainMode CookieDomainMode
CookieTestURL string
}
type CookieDomainMode int
@@ -44,7 +43,7 @@ func RunProviderContract(t *testing.T, provider enginewebview.Provider, opts Con
}
waitForProviderAvailable(t, provider, timeout)
page, server, testURL := openTestPage(t, provider, enginewebview.OpenOptions{
page, _, testURL := openTestPage(t, provider, enginewebview.OpenOptions{
Headless: true,
Title: "Gopeed WebView Integration",
Width: 640,
@@ -155,11 +154,9 @@ func RunProviderContract(t *testing.T, provider enginewebview.Provider, opts Con
t.Fatal("expected missing selector wait to time out")
}
cookieTestURL := opts.CookieTestURL
if cookieTestURL == "" {
cookieTestURL = testURL
}
runCookieLifecycleAssertions(t, page, server, cookieTestURL, opts.CookieDomainMode)
// Keep cookie navigation on the same local fixture as the DOM assertions:
// this contract must not depend on public DNS, TLS, or site availability.
runCookieLifecycleAssertions(t, page, testURL, opts.CookieDomainMode)
}
func waitForProviderAvailable(t *testing.T, provider enginewebview.Provider, timeout time.Duration) {
@@ -270,7 +267,7 @@ func openTestPage(t *testing.T, provider enginewebview.Provider, opts enginewebv
return page, server, server.URL
}
func runCookieLifecycleAssertions(t *testing.T, page *enginewebview.PageHandle, _ *httptest.Server, testURL string, domainMode CookieDomainMode) {
func runCookieLifecycleAssertions(t *testing.T, page *enginewebview.PageHandle, testURL string, domainMode CookieDomainMode) {
t.Helper()
parsedURL, err := url.Parse(testURL)
@@ -26,7 +26,6 @@ func TestProviderContract(t *testing.T) {
})
integrationtest.RunProviderContract(t, provider, integrationtest.ContractOptions{
CookieDomainMode: integrationtest.CookieDomainModeRequired,
CookieTestURL: "https://example.com/",
})
}
+26
View File
@@ -0,0 +1,26 @@
package download
import (
"os"
"path/filepath"
"runtime/debug"
"testing"
)
func TestLogPanicPreservesPreviousCrash(t *testing.T) {
dir := t.TempDir()
path := filepath.Join(dir, "crash.log")
const previous = "previous crash traceback\n"
if err := os.WriteFile(path, []byte(previous), 0600); err != nil {
t.Fatal(err)
}
logPanic(dir)
t.Cleanup(func() { _ = debug.SetCrashOutput(nil, debug.CrashOptions{}) })
data, err := os.ReadFile(path)
if err != nil {
t.Fatal(err)
}
if string(data) != previous {
t.Fatalf("previous crash was overwritten: %q", data)
}
}
+23 -9
View File
@@ -345,7 +345,12 @@ func (d *Downloader) Setup() error {
}
current := task.fetcher.Progress().TotalDownloaded()
delta := current - task.Progress.Downloaded
delta := max(0, current-task.Progress.Downloaded)
if counter, ok := task.fetcher.(fetcher.DownloadCounter); ok && task.Status == base.DownloadStatusRunning {
received := counter.ReceivedBytes()
delta = received - task.receivedBytes
task.receivedBytes = received
}
if task.Meta != nil && task.Meta.Req != nil {
if producer := d.blob.Production(task.Meta.Req.URL); producer != nil {
downloaded, received := producer.Progress()
@@ -353,19 +358,20 @@ func (d *Downloader) Setup() error {
task.producer = producer
task.producerReceived = 0
task.speedArr = nil
task.downloadSampledAt = time.Time{}
}
current = downloaded
delta = max(0, received-task.producerReceived)
task.producerReceived = received
}
}
tick := float64(d.cfg.RefreshInterval) / 1000
downloadDataChanged := false
uploading := task.Uploading
if task.Status == base.DownloadStatusRunning {
downloadDataChanged = current != task.Progress.Downloaded
task.Progress.Used = task.timer.Used()
task.Progress.Speed = task.updateSpeed(max(0, delta), tick)
elapsed := sampleSeconds(&task.downloadSampledAt, time.Now())
task.Progress.Speed = task.updateSpeed(delta, elapsed)
task.Progress.Downloaded = current
}
@@ -374,7 +380,8 @@ func (d *Downloader) Setup() error {
uploader := task.fetcher.(fetcher.Uploader)
currentUploaded := uploader.UploadedBytes()
uploadDataChanged = currentUploaded != task.Progress.Uploaded
task.Progress.UploadSpeed = task.updateUploadSpeed(currentUploaded-task.Progress.Uploaded, tick)
elapsed := sampleSeconds(&task.uploadSampledAt, time.Now())
task.Progress.UploadSpeed = task.updateUploadSpeed(currentUploaded-task.Progress.Uploaded, elapsed)
task.Progress.Uploaded = currentUploaded
}
task.statusLock.Unlock()
@@ -2405,7 +2412,9 @@ func (d *Downloader) doStart(task *Task) (err error) {
return err
}
}
d.triggerOnStart(task)
if err := d.triggerOnStart(task); err != nil {
return err
}
if !d.taskIsRunningGeneration(task, generation) {
return nil
}
@@ -2453,6 +2462,8 @@ func (d *Downloader) doStart(task *Task) (err error) {
return nil
}
task.Progress.Speed = 0
task.speedArr = nil
task.downloadSampledAt = time.Time{}
task.timer.Start()
task.statusLock.Unlock()
if err := task.fetcher.Start(); err != nil {
@@ -2650,11 +2661,12 @@ func logPanic(logDir string) {
if err := util.CreateDirIfNotExist(logDir); err != nil {
return
}
f, err := os.Create(filepath.Join(logDir, "crash.log"))
f, err := os.OpenFile(filepath.Join(logDir, "crash.log"), os.O_CREATE|os.O_WRONLY|os.O_APPEND, 0600)
if err != nil {
return
}
debug.SetCrashOutput(f, debug.CrashOptions{})
defer f.Close() // SetCrashOutput duplicates the descriptor.
_ = debug.SetCrashOutput(f, debug.CrashOptions{})
}
func (d *Downloader) assignFetcherManager(task *Task) error {
@@ -3061,8 +3073,10 @@ func initTask(task *Task) {
task.lock = &sync.Mutex{}
task.statsLock = &sync.Mutex{}
task.blobRefLock = &sync.Mutex{}
task.speedArr = make([]int64, 0)
task.uploadSpeedArr = make([]int64, 0)
task.speedArr = nil
task.uploadSpeedArr = nil
task.downloadSampledAt = time.Time{}
task.uploadSampledAt = time.Time{}
}
func (d *Downloader) releaseBlobTask(task *Task) {
+47
View File
@@ -0,0 +1,47 @@
package engine
import (
"context"
"errors"
"github.com/dop251/goja"
"testing"
"time"
)
func TestCloseUnblocksPendingPromise(t *testing.T) {
e := NewEngine(nil)
entered := make(chan struct{})
_, err := e.runOnLoop(func(r *goja.Runtime) (goja.Value, error) {
return goja.Undefined(), r.Set("entered", func() { close(entered) })
})
if err != nil {
t.Fatal(err)
}
done := make(chan error, 1)
go func() { _, err := e.RunString(`entered(); new Promise(()=>{});`); done <- err }()
select {
case <-entered:
case <-time.After(time.Second):
t.Fatal("script did not enter")
}
e.Close()
select {
case err := <-done:
if err == nil {
t.Fatal("closed execution returned success")
}
case <-time.After(300 * time.Millisecond):
t.Fatal("Engine.Close left the Promise waiter blocked")
}
}
func TestExecutionDeadlineInterruptsSynchronousScript(t *testing.T) {
e := NewEngine(nil)
defer e.Close()
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Millisecond)
defer cancel()
_, err := e.RunStringContext(ctx, `while (true) {}`)
if !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("got %v, want deadline exceeded", err)
}
}
+49 -6
View File
@@ -1,6 +1,7 @@
package engine
import (
"context"
_ "embed"
"errors"
"fmt"
@@ -26,6 +27,7 @@ var polyfillScript string
type Engine struct {
loop *eventloop.EventLoop
done chan struct{}
closeOnce sync.Once
cleanupMu sync.Mutex
@@ -39,7 +41,13 @@ type JSFunction func(goja.FunctionCall) goja.Value
// RunString executes the script and returns the go type value
// if script result is promise, it will be resolved
func (e *Engine) RunString(script string) (value any, err error) {
return e.runOnLoop(func(runtime *goja.Runtime) (goja.Value, error) {
return e.RunStringContext(context.Background(), script)
}
// RunStringContext bounds an execution, including awaited promises. Cancelling
// an execution closes its engine so outstanding operations cannot outlive it.
func (e *Engine) RunStringContext(ctx context.Context, script string) (any, error) {
return e.runOnLoopContext(ctx, func(runtime *goja.Runtime) (goja.Value, error) {
return runtime.RunString(script)
})
}
@@ -47,7 +55,11 @@ func (e *Engine) RunString(script string) (value any, err error) {
// CallFunction calls the function and returns the go type value
// if function result is promise, it will be resolved
func (e *Engine) CallFunction(fn any, args ...any) (value any, err error) {
return e.runOnLoop(func(runtime *goja.Runtime) (goja.Value, error) {
return e.CallFunctionContext(context.Background(), fn, args...)
}
func (e *Engine) CallFunctionContext(ctx context.Context, fn any, args ...any) (value any, err error) {
return e.runOnLoopContext(ctx, func(runtime *goja.Runtime) (goja.Value, error) {
var jsArgs []goja.Value
for _, arg := range args {
jsArgs = append(jsArgs, runtime.ToValue(arg))
@@ -69,6 +81,20 @@ func (e *Engine) CallFunction(fn any, args ...any) (value any, err error) {
}
func (e *Engine) runOnLoop(fn func(runtime *goja.Runtime) (goja.Value, error)) (any, error) {
return e.runOnLoopContext(context.Background(), fn)
}
var ErrClosed = errors.New("engine closed")
func (e *Engine) runOnLoopContext(ctx context.Context, fn func(runtime *goja.Runtime) (goja.Value, error)) (any, error) {
if err := ctx.Err(); err != nil {
return nil, err
}
select {
case <-e.done:
return nil, ErrClosed
default:
}
type result struct {
value any
err error
@@ -132,15 +158,31 @@ func (e *Engine) runOnLoop(fn func(runtime *goja.Runtime) (goja.Value, error)) (
if !ok {
return nil, errors.New("engine loop terminated")
}
res := <-ch
if res.err != nil {
return nil, res.err
select {
case res := <-ch:
return res.value, res.err
case <-ctx.Done():
// Interrupt synchronous JS too; a select alone only stops Promise waiting.
e.Close()
return nil, ctx.Err()
case <-e.done:
return nil, ErrClosed
}
return res.value, nil
}
func (e *Engine) Close() {
e.closeOnce.Do(func() {
close(e.done)
e.Runtime.Interrupt(ErrClosed)
// Wait for interrupted JS to leave the loop, then allow cleanup hooks
// to execute their JS cancellation callbacks before terminating it.
cleared := make(chan struct{})
if e.loop.RunOnLoop(func(runtime *goja.Runtime) {
runtime.ClearInterrupt()
close(cleared)
}) {
<-cleared
}
e.cleanupMu.Lock()
cleanups := e.cleanups
e.cleanups = nil
@@ -177,6 +219,7 @@ func NewEngine(cfg *Config) *Engine {
loop := eventloop.NewEventLoop()
engine := &Engine{
loop: loop,
done: make(chan struct{}),
}
loop.Start()
done := make(chan struct{})
+9
View File
@@ -174,3 +174,12 @@ compilation memory, host buffers and JS allocations are additional to the
linear-memory limit. Playback support depends on the selected codecs, not just
the `.mp4` suffix. Platform execution and performance must be tested on target
devices, especially mobile/32-bit devices.
On 64-bit Windows, instance linear memory uses a reserved virtual address range with
pages committed as it grows. Module completion, errors and cancellation release
this range directly through `VirtualFree`, rather than waiting for Go's garbage
collector to reclaim large buffers. Growing an instance preserves its address
without copying the previous buffer. The 512 MiB limit is unchanged; reservation
does not commit that entire amount. Shared compilation remains resident after
the first merge, so process memory is not expected to return exactly to the
pre-FFmpeg baseline. This does not diagnose unrelated idle memory growth.
+32
View File
@@ -0,0 +1,32 @@
package ffmpeg
import (
"context"
"sync"
"github.com/tetratelabs/wazero/experimental"
)
// The runtime does not always invoke Free when a module's start function traps.
// Keep host ownership until execution returns, including failed instantiation.
func withOwnedMemory(ctx context.Context, allocator experimental.MemoryAllocator) (context.Context, func()) {
var memories []*ownedLinearMemory
ctx = experimental.WithMemoryAllocator(ctx, experimental.MemoryAllocatorFunc(
func(cap, max uint64) experimental.LinearMemory {
memory := &ownedLinearMemory{LinearMemory: allocator.Allocate(cap, max)}
memories = append(memories, memory)
return memory
}))
return ctx, func() {
for _, memory := range memories {
memory.Free()
}
}
}
type ownedLinearMemory struct {
experimental.LinearMemory
once sync.Once
}
func (m *ownedLinearMemory) Free() { m.once.Do(m.LinearMemory.Free) }
@@ -0,0 +1,7 @@
//go:build !windows || (!amd64 && !arm64)
package ffmpeg
import "context"
func withLinearMemory(ctx context.Context) (context.Context, func()) { return ctx, func() {} }
+88
View File
@@ -0,0 +1,88 @@
package ffmpeg
import (
"context"
"encoding/hex"
"testing"
"time"
"github.com/tetratelabs/wazero"
"github.com/tetratelabs/wazero/experimental"
)
// Verify the pinned runtime's allocator lifecycle on every development host:
// host-owned buffers must not be released before execution ends, or retained
// after failed instantiation, traps or cancellation.
func TestLinearMemoryModuleLifecycle(t *testing.T) {
testLinearMemoryModuleLifecycle(t, func(_, _ uint64) experimental.LinearMemory {
return &testLinearMemory{}
})
}
type testLinearMemory struct{ buffer []byte }
func (m *testLinearMemory) Reallocate(size uint64) []byte {
m.buffer = append(m.buffer, make([]byte, int(size)-len(m.buffer))...)
return m.buffer
}
func (m *testLinearMemory) Free() { m.buffer = nil }
type trackedLinearMemory struct {
experimental.LinearMemory
freeCount int
}
func (m *trackedLinearMemory) Free() {
m.freeCount++
m.LinearMemory.Free()
}
func testLinearMemoryModuleLifecycle(t *testing.T, allocate experimental.MemoryAllocatorFunc) {
t.Helper()
for _, tt := range []struct {
name string
// A module with one memory (min=1/max=2 pages), optionally a start
// function that traps or loops until its context is cancelled.
wasm string
wantErr bool
}{
{"close", "0061736d01000000050401010102", false},
{"trap", "0061736d01000000010401600000030201000504010101020801000a05010300000b", true},
{"cancel", "0061736d01000000010401600000030201000504010101020801000a0901070003400c000b0b", true},
} {
t.Run(tt.name, func(t *testing.T) {
ctx, cancel := context.WithTimeout(context.Background(), time.Second)
defer cancel()
var memory *trackedLinearMemory
ctx, release := withOwnedMemory(ctx, experimental.MemoryAllocatorFunc(
func(_, max uint64) experimental.LinearMemory {
memory = &trackedLinearMemory{LinearMemory: allocate(0, max)}
return memory
}))
defer release()
rt := wazero.NewRuntimeWithConfig(ctx, wazero.NewRuntimeConfig().WithCloseOnContextDone(true))
defer rt.Close(context.Background())
binary, err := hex.DecodeString(tt.wasm)
if err != nil {
t.Fatal(err)
}
module, err := rt.Instantiate(ctx, binary)
if (err != nil) != tt.wantErr {
t.Fatalf("instantiate: %v", err)
}
if !tt.wantErr {
if memory == nil || memory.freeCount != 0 {
t.Fatal("module did not allocate native memory")
}
if err := module.Close(ctx); err != nil {
t.Fatal(err)
}
}
release()
release()
if memory == nil || memory.freeCount != 1 {
t.Fatal("host did not release memory exactly once")
}
})
}
}
@@ -0,0 +1,70 @@
//go:build windows && (amd64 || arm64)
package ffmpeg
import (
"context"
"fmt"
"unsafe"
"github.com/tetratelabs/wazero/experimental"
"golang.org/x/sys/windows"
)
// Keep large, short-lived WASM buffers out of the Go heap. Closing a module
// releases its committed pages immediately, without waiting for GC/scavenging.
func withLinearMemory(ctx context.Context) (context.Context, func()) {
return withOwnedMemory(ctx, experimental.MemoryAllocatorFunc(
func(_, max uint64) experimental.LinearMemory {
return &windowsLinearMemory{max: max}
}))
}
// Reserve the address range once, committing zero-filled pages on demand. This
// avoids both copying the previous buffer and temporarily retaining two large
// Go allocations when memory.grow runs. The backing address never changes.
type windowsLinearMemory struct {
address uintptr
max uint64
size uint64
freed bool
}
func (m *windowsLinearMemory) Reallocate(size uint64) []byte {
if m.freed || size > m.max || m.max > uint64(^uint(0)>>1) {
return nil
}
if size == 0 {
return []byte{}
}
if m.address == 0 {
address, err := windows.VirtualAlloc(0, uintptr(m.max), windows.MEM_RESERVE, windows.PAGE_READWRITE)
if err != nil {
panic(fmt.Errorf("ffmpeg: reserve WASM memory: %w", err))
}
m.address = address
}
if size > m.size {
_, err := windows.VirtualAlloc(m.address, uintptr(size), windows.MEM_COMMIT, windows.PAGE_READWRITE)
if err != nil {
if m.size == 0 {
// Instantiation cannot proceed without the minimum memory. Wazero
// has not taken ownership yet, so release the reservation here.
m.Free()
panic(fmt.Errorf("ffmpeg: commit WASM memory: %w", err))
}
return nil // memory.grow reports failure to the guest.
}
m.size = size
}
return unsafe.Slice((*byte)(unsafe.Pointer(m.address)), int(size))
}
func (m *windowsLinearMemory) Free() {
if m.address != 0 {
_ = windows.VirtualFree(m.address, 0, windows.MEM_RELEASE)
m.address = 0
}
m.size = 0
m.freed = true
}
@@ -0,0 +1,54 @@
//go:build windows && (amd64 || arm64)
package ffmpeg
import (
"testing"
"unsafe"
"github.com/tetratelabs/wazero/experimental"
"golang.org/x/sys/windows"
)
func TestWindowsLinearMemoryGrowthAndRelease(t *testing.T) {
const page = 65536
m := &windowsLinearMemory{max: 4 * page}
defer m.Free()
first := m.Reallocate(page)
if len(first) != page {
t.Fatalf("initial allocation: %d", len(first))
}
first[0], first[page-1] = 17, 29
address := m.address
grown := m.Reallocate(3 * page)
if unsafe.SliceData(first) != unsafe.SliceData(grown) || grown[0] != 17 || grown[page-1] != 29 {
t.Fatal("growing memory moved or lost existing bytes")
}
for _, b := range grown[page:] {
if b != 0 {
t.Fatal("newly committed memory is not zero-filled")
}
}
if m.Reallocate(5*page) != nil || m.address != address || m.size != 3*page {
t.Fatal("failed growth changed the existing allocation")
}
m.Free()
m.Free()
if m.address != 0 || m.Reallocate(page) != nil {
t.Fatal("freed memory can still be used")
}
var info windows.MemoryBasicInformation
if err := windows.VirtualQuery(address, &info, unsafe.Sizeof(info)); err != nil {
t.Fatal(err)
}
const memFree = 0x10000 // MEMORY_BASIC_INFORMATION.State: MEM_FREE.
if info.State != memFree {
t.Fatalf("pages remain reserved after Free: state=%x", info.State)
}
}
func TestWindowsLinearMemoryModuleLifecycle(t *testing.T) {
testLinearMemoryModuleLifecycle(t, func(_, max uint64) experimental.LinearMemory {
return &windowsLinearMemory{max: max}
})
}
+3 -1
View File
@@ -151,7 +151,9 @@ func RunAcquired(ctx context.Context, video, audio Input, out io.Writer, format
files := &inputFS{files: map[string]Input{"video": video, "audio": audio}}
fsconfig := wazero.NewFSConfig().(sysfs.FSConfig).WithSysFSMount(files, "/inputs")
logs := &tailLog{}
rc, err := wasm.Run(ctx, shared.runtime, shared.module, wasm.Args{
runCtx, freeMemory := withLinearMemory(ctx)
defer freeMemory()
rc, err := wasm.Run(runCtx, shared.runtime, shared.module, wasm.Args{
Name: "ffmpeg", Args: args, Stdout: out, Stderr: logs,
Config: func(c wazero.ModuleConfig) wazero.ModuleConfig { return c.WithFSConfig(fsconfig) },
})
+15 -4
View File
@@ -3,6 +3,7 @@ package download
import (
"context"
"encoding/json"
"errors"
"fmt"
"io"
"net/url"
@@ -327,8 +328,8 @@ func (d *Downloader) triggerOnResolve(req *base.Request) (res *base.Resource, er
return
}
func (d *Downloader) triggerOnStart(task *Task) {
doTrigger(d,
func (d *Downloader) triggerOnStart(task *Task) error {
return doTrigger(d,
EventOnStart,
task.Meta.Req,
&OnStartContext{
@@ -413,6 +414,13 @@ func doTrigger[T any](d *Downloader, event ActivationEvent, req *base.Request, c
}
engine, session := d.newExtensionEngine()
defer session.CloseIfIdle()
eventCtx, cancelEvent := context.WithTimeout(d.backgroundCtx, d.cfg.ExtensionEventTimeout)
defer cancelEvent()
defer func() {
if eventCtx.Err() != nil {
session.Close()
}
}()
gopeed.Runtime = &InstanceRuntime{
WebView: d.newExtensionWebViewRuntime(session, ext.buildIdentity()),
}
@@ -421,13 +429,13 @@ func doTrigger[T any](d *Downloader, event ActivationEvent, req *base.Request, c
gopeed.Logger.logger.Error().Err(err).Msgf("[%s] engine inject failed", ext.buildIdentity())
return
}
_, err = engine.RunString(string(scriptBuf))
_, err = engine.RunStringContext(eventCtx, string(scriptBuf))
if err != nil {
gopeed.Logger.logger.Error().Err(err).Msgf("[%s] run script failed", ext.buildIdentity())
return
}
if fn, ok := gopeed.Events[event]; ok {
_, err = engine.CallFunction(fn, ctx)
_, err = engine.CallFunctionContext(eventCtx, fn, ctx)
if err != nil {
gopeed.Logger.logger.Error().Err(err).Msgf("[%s] call function failed: %s", ext.buildIdentity(), event)
return
@@ -437,6 +445,9 @@ func doTrigger[T any](d *Downloader, event ActivationEvent, req *base.Request, c
}
}
}()
if errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
return err
}
}
}
}
+149
View File
@@ -0,0 +1,149 @@
package download
import (
"context"
"errors"
"fmt"
"net/http"
"net/http/httptest"
"os"
"path/filepath"
"sync"
"testing"
"time"
"github.com/GopeedLab/gopeed/internal/fetcher"
"github.com/GopeedLab/gopeed/pkg/base"
)
func installCancellationExtension(t *testing.T, d *Downloader, event, script string) {
t.Helper()
dir := t.TempDir()
manifest := fmt.Sprintf(`{"name":"cancel-%s","author":"test","title":"Cancel","version":"1.0.0","scripts":[{"event":%q,"match":{"urls":["*://example.com/*"]},"entry":"index.js"}]}`, event, event)
for name, body := range map[string]string{"manifest.json": manifest, "index.js": script} {
if err := os.WriteFile(filepath.Join(dir, name), []byte(body), 0600); err != nil {
t.Fatal(err)
}
}
if _, err := d.InstallExtensionByFolder(dir, true); err != nil {
t.Fatal(err)
}
}
func newCancellationDownloader(t *testing.T) *Downloader {
t.Helper()
d := NewDownloader(&DownloaderConfig{Storage: NewMemStorage(), StorageDir: t.TempDir(), TempDir: t.TempDir(), FetchManagers: []fetcher.FetcherManager{&generationTestManager{holdOpen: true}}})
if err := d.Setup(); err != nil {
t.Fatal(err)
}
return d
}
func TestResolveExtensionStopsOnShutdown(t *testing.T) {
entered, release := make(chan struct{}), make(chan struct{})
var once sync.Once
defer once.Do(func() { close(release) })
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
close(entered)
select {
case <-release:
case <-r.Context().Done():
}
w.Write([]byte("ok"))
}))
defer server.Close()
d := newCancellationDownloader(t)
defer d.Close()
installCancellationExtension(t, d, "onResolve", fmt.Sprintf(`gopeed.events.onResolve(async ctx=>{await fetch(%q);});`, server.URL))
done := make(chan error, 1)
go func() { _, err := d.Resolve(&base.Request{URL: "https://example.com/cancel"}); done <- err }()
select {
case <-entered:
case <-time.After(2 * time.Second):
t.Fatal("handler did not enter")
}
if err := d.Close(); err != nil {
t.Fatal(err)
}
select {
case err := <-done:
if !errors.Is(err, context.Canceled) {
t.Errorf("got %v, want cancellation", err)
}
case <-time.After(300 * time.Millisecond):
once.Do(func() { close(release) })
t.Fatal("shutdown did not cancel extension Resolve")
}
}
func TestOnErrorCancellationReleasesTaskAPIs(t *testing.T) {
entered, release := make(chan struct{}), make(chan struct{})
var once sync.Once
defer once.Do(func() { close(release) })
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
close(entered)
select {
case <-release:
case <-r.Context().Done():
}
w.Write([]byte("ok"))
}))
defer server.Close()
d := newCancellationDownloader(t)
defer d.Close()
parent, cancel := context.WithCancel(d.backgroundCtx)
d.backgroundCtx = parent
defer cancel()
installCancellationExtension(t, d, "onError", fmt.Sprintf(`gopeed.events.onError(async ctx=>{await fetch(%q);});`, server.URL))
d.cfg.MaxRunning = 0
id, err := d.CreateDirect(&base.Request{URL: "generation://example.com/task"}, &base.Options{Path: t.TempDir()})
if err != nil {
t.Fatal(err)
}
task := d.GetTask(id)
task.statusLock.Lock()
task.Status = base.DownloadStatusRunning
task.statusLock.Unlock()
handled := make(chan struct{})
go func() { d.doOnError(task, errors.New("source failed")); close(handled) }()
select {
case <-entered:
case <-time.After(2 * time.Second):
t.Fatal("onError did not enter")
}
status, stats := make(chan struct{}), make(chan struct{})
go func() { d.RuntimeStatus(id); close(status) }()
go func() { d.Stats(id); close(stats) }()
cancel()
for name, ch := range map[string]<-chan struct{}{"onError": handled, "RuntimeStatus": status, "Stats": stats} {
select {
case <-ch:
case <-time.After(300 * time.Millisecond):
once.Do(func() { close(release) })
<-handled
t.Fatalf("%s stayed blocked after cancellation", name)
}
}
}
func TestExtensionEventDeadlineRejectsUnsettledResolve(t *testing.T) {
d := newCancellationDownloader(t)
defer d.Close()
d.cfg.ExtensionEventTimeout = 30 * time.Millisecond
installCancellationExtension(t, d, "onResolve", `gopeed.events.onResolve(async ctx=>{await new Promise(()=>{});});`)
_, err := d.Resolve(&base.Request{URL: "https://example.com/timeout"})
if !errors.Is(err, context.DeadlineExceeded) {
t.Fatalf("got %v, want event deadline exceeded", err)
}
}
func TestExtensionStartDeadlinePreventsFetcherStart(t *testing.T) {
d := newCancellationDownloader(t)
defer d.Close()
d.cfg.ExtensionEventTimeout = 30 * time.Millisecond
installCancellationExtension(t, d, "onStart", `gopeed.events.onStart(async ctx=>{await new Promise(()=>{});});`)
id, err := d.CreateDirect(&base.Request{URL: "generation://example.com/timeout"}, &base.Options{Path: t.TempDir()})
if err != nil {
t.Fatal(err)
}
waitForTaskStatus(t, d, id, base.DownloadStatusError, 2*time.Second)
if starts := d.cfg.FetchManagers[0].(*generationTestManager).starts.Load(); starts != 0 {
t.Fatalf("timed out onStart still started fetcher %d times", starts)
}
}
+15
View File
@@ -59,6 +59,21 @@ func (s *engineSession) Release() {
}
}
// Close releases retained resources when the owning event is cancelled.
func (s *engineSession) Close() {
s.mu.Lock()
if s.closed {
s.mu.Unlock()
return
}
s.closed = true
s.mu.Unlock()
s.runClosers()
if s.engine != nil {
s.engine.Close()
}
}
func (s *engineSession) CloseIfIdle() {
s.mu.Lock()
shouldClose := s.refs == 0 && !s.closed
+54 -31
View File
@@ -75,20 +75,23 @@ type Task struct {
producer production.Source
producerReceived int64
fetcherManager fetcher.FetcherManager
fetcher fetcher.Fetcher
timer *util.Timer
statusLock *sync.Mutex
lock *sync.Mutex
statsLock *sync.Mutex
statsLoaded bool
stats *fetcher.Stats
statsSnapshot json.RawMessage
blobRefLock *sync.Mutex
blobURL string
runGeneration uint64
speedArr []int64
uploadSpeedArr []int64
fetcherManager fetcher.FetcherManager
fetcher fetcher.Fetcher
timer *util.Timer
statusLock *sync.Mutex
lock *sync.Mutex
statsLock *sync.Mutex
statsLoaded bool
stats *fetcher.Stats
statsSnapshot json.RawMessage
blobRefLock *sync.Mutex
blobURL string
runGeneration uint64
speedArr []speedSample
uploadSpeedArr []speedSample
downloadSampledAt time.Time
uploadSampledAt time.Time
receivedBytes int64
}
// TaskRuntimeStatus is the lightweight, frequently refreshed view of a task.
@@ -195,7 +198,21 @@ func (t *Task) updateUploadSpeed(downloaded int64, usedTime float64) int64 {
return calcSpeed(&t.uploadSpeedArr, downloaded, usedTime)
}
func calcSpeed(speedArr *[]int64, downloaded int64, usedTime float64) int64 {
type speedSample struct {
bytes int64
seconds float64
}
func sampleSeconds(lastSample *time.Time, now time.Time) float64 {
previous := *lastSample
*lastSample = now
if previous.IsZero() {
return 0
}
return now.Sub(previous).Seconds()
}
func calcSpeed(speedArr *[]speedSample, downloaded int64, usedTime float64) int64 {
if usedTime <= 0 {
return 0
}
@@ -204,18 +221,18 @@ func calcSpeed(speedArr *[]int64, downloaded int64, usedTime float64) int64 {
return 0
}
*speedArr = append(*speedArr, downloaded)
// Record last 5 seconds of download speed to calculate the average speed
if len(*speedArr) > int(5.0/usedTime) {
*speedArr = (*speedArr)[1:]
}
*speedArr = append(*speedArr, speedSample{bytes: downloaded, seconds: usedTime})
var total int64
for _, v := range *speedArr {
total += v
var elapsed float64
first := len(*speedArr)
for first > 0 && elapsed < 5 {
first--
sample := (*speedArr)[first]
total += sample.bytes
elapsed += sample.seconds
}
return int64(float64(total) / float64(len(*speedArr)) / usedTime)
*speedArr = (*speedArr)[first:]
return int64(float64(total) / elapsed)
}
type TaskFilter struct {
@@ -232,12 +249,15 @@ type DownloaderConfig struct {
Controller *controller.Controller
FetchManagers []fetcher.FetcherManager
RefreshInterval int // RefreshInterval time duration to refresh task progress(ms)
Storage Storage
StorageDir string
TempDir string
WhiteDownloadDirs []string
WebViewProvider enginewebview.Provider
// ExtensionEventTimeout bounds one script load and its awaited event handler.
// Zero uses one minute. Embedders may increase it for long-running handlers.
ExtensionEventTimeout time.Duration
RefreshInterval int // RefreshInterval time duration to refresh task progress(ms)
Storage Storage
StorageDir string
TempDir string
WhiteDownloadDirs []string
WebViewProvider enginewebview.Provider
ProductionMode bool
@@ -260,6 +280,9 @@ func (cfg *DownloaderConfig) Init() *DownloaderConfig {
new(ed2k.FetcherManager),
}
}
if cfg.ExtensionEventTimeout <= 0 {
cfg.ExtensionEventTimeout = time.Minute
}
if cfg.RefreshInterval == 0 {
cfg.RefreshInterval = 350
}
+39 -2
View File
@@ -1,9 +1,32 @@
package download
import "testing"
import (
"testing"
"time"
)
func TestTransferSpeedWithUnevenSamples(t *testing.T) {
var task Task
for _, sample := range []struct {
bytes int64
seconds float64
}{
{1000, 1},
{2000, 2},
{4000, 4},
{6000, 6},
} {
if got := task.updateSpeed(sample.bytes, sample.seconds); got != 1000 {
t.Errorf("download speed after %v seconds = %d, want 1000", sample.seconds, got)
}
if got := task.updateUploadSpeed(sample.bytes, sample.seconds); got != 1000 {
t.Errorf("upload speed after %v seconds = %d, want 1000", sample.seconds, got)
}
}
}
func TestCalcSpeedResetOnRollback(t *testing.T) {
speedArr := []int64{1024, 2048, 4096}
speedArr := []speedSample{{1024, 1}, {2048, 1}, {4096, 1}}
if got := calcSpeed(&speedArr, -512, 1); got != 0 {
t.Fatalf("calcSpeed() = %d, want 0 after rollback", got)
@@ -16,3 +39,17 @@ func TestCalcSpeedResetOnRollback(t *testing.T) {
t.Fatalf("calcSpeed() = %d, want 1024 after reset", got)
}
}
func TestSampleSecondsUsesThePreviousSuccessfulSample(t *testing.T) {
start := time.Now()
var last time.Time
if got := sampleSeconds(&last, start); got != 0 {
t.Fatalf("first sample elapsed = %v, want 0", got)
}
if got := sampleSeconds(&last, start.Add(1500*time.Millisecond)); got != 1.5 {
t.Fatalf("elapsed after a delayed sample = %v, want 1.5", got)
}
if got := sampleSeconds(&last, start.Add(2*time.Second)); got != 0.5 {
t.Fatalf("elapsed after the next sample = %v, want 0.5", got)
}
}
+7 -2
View File
@@ -27,6 +27,10 @@ type ScriptData struct {
// ScriptPayload contains the task data
type ScriptPayload struct {
Task *Task
// TaskName is computed before Task is cloned. The clone has no fetcher
// manager, so for an unresolved task without Options.Name its Name()
// would return "unknown".
TaskName string
}
// getScriptPaths extracts script paths from config
@@ -103,7 +107,7 @@ func (d *Downloader) executeScriptAtPath(scriptPath string, data *ScriptData) er
cmd.Env = append(os.Environ(),
fmt.Sprintf("GOPEED_EVENT=%s", data.Event),
fmt.Sprintf("GOPEED_TASK_ID=%s", data.Payload.Task.ID),
fmt.Sprintf("GOPEED_TASK_NAME=%s", data.Payload.Task.Name()),
fmt.Sprintf("GOPEED_TASK_NAME=%s", data.Payload.TaskName),
fmt.Sprintf("GOPEED_TASK_STATUS=%s", data.Payload.Task.Status),
)
@@ -137,7 +141,8 @@ func (d *Downloader) triggerScripts(event ScriptEvent, task *Task, err error) {
Event: event,
Time: time.Now().UnixMilli(),
Payload: &ScriptPayload{
Task: task.clone(),
Task: task.clone(),
TaskName: task.Name(),
},
}
+67
View File
@@ -3,6 +3,8 @@ package download
import (
"os"
"path/filepath"
"runtime"
"strings"
"testing"
"time"
@@ -126,6 +128,46 @@ func TestScript_ExecuteScriptAtPath_NonExistentFile(t *testing.T) {
})
}
func TestScript_TaskNameOfUnresolvedTask(t *testing.T) {
scriptPath := getTestScriptPath(t, envDumpScriptName())
ensureScriptExecutable(t, scriptPath)
outputFile := filepath.Join(t.TempDir(), "env_output.txt")
setupScriptTest(t, func(downloader *Downloader) {
cfg, _ := downloader.GetConfig()
cfg.Script = &base.ScriptConfig{
Enable: true,
Paths: []string{scriptPath},
}
downloader.PutConfig(cfg)
// Without Options.Name and a resource, the name comes from the fetcher
// manager, which the cloned task in the script data does not have.
url := "http://127.0.0.1/error.bin"
fm, err := downloader.parseFm(url)
if err != nil {
t.Fatal(err)
}
task := NewTask()
task.Meta = &fetcher.FetcherMeta{
Req: &base.Request{URL: url},
Opts: &base.Options{Path: t.TempDir()},
}
task.fetcherManager = fm
if name := task.Name(); name != "error.bin" {
t.Fatalf("Expected task name error.bin, got %q", name)
}
t.Setenv("GOPEED_TEST_OUTPUT_FILE", outputFile)
downloader.triggerScripts(ScriptEventDownloadDone, task, nil)
output := waitForFileContains(t, outputFile, "GOPEED_TASK_PATH=", 3*time.Second)
if !strings.Contains(output, "GOPEED_TASK_NAME=error.bin") {
t.Errorf("Expected GOPEED_TASK_NAME=error.bin in output, got: %s", output)
}
})
}
func createDownloadDoneTask(t *testing.T, downloadDir, fileName string) (*Task, string) {
t.Helper()
content := []byte("downloaded file")
@@ -171,6 +213,31 @@ func waitForFile(t *testing.T, path string, timeout time.Duration) {
t.Fatalf("Timeout waiting for file: %s", path)
}
// waitForFileContains waits until the file contains want and returns its content.
// Test scripts write the file line by line, so waiting for the file to exist is not enough.
func waitForFileContains(t *testing.T, path, want string, timeout time.Duration) string {
t.Helper()
deadline := time.Now().Add(timeout)
for time.Now().Before(deadline) {
if content, err := os.ReadFile(path); err == nil && strings.Contains(string(content), want) {
return string(content)
}
time.Sleep(50 * time.Millisecond)
}
content, _ := os.ReadFile(path)
t.Fatalf("Timeout waiting for %q in file %s, got: %q", want, path, content)
return ""
}
// envDumpScriptName returns the test script that writes the GOPEED_* variables
// into the file named by GOPEED_TEST_OUTPUT_FILE.
func envDumpScriptName() string {
if runtime.GOOS == "windows" {
return "env_dump.bat"
}
return "env_dump.sh"
}
func getTestScriptPath(t *testing.T, name string) string {
t.Helper()
path := filepath.Join("testdata", "scripts", name)
+2 -7
View File
@@ -91,13 +91,8 @@ func TestScript_EnvironmentVariables(t *testing.T) {
t.Setenv("GOPEED_TEST_OUTPUT_FILE", outputFile)
downloader.triggerScripts(ScriptEventDownloadDone, task, nil)
waitForFile(t, outputFile, 3*time.Second)
content, err := os.ReadFile(outputFile)
if err != nil {
t.Fatalf("Failed to read output file: %v", err)
}
output := string(content)
// GOPEED_TASK_PATH is the last line the script writes.
output := waitForFileContains(t, outputFile, "GOPEED_TASK_PATH=", 3*time.Second)
if !strings.Contains(output, "GOPEED_EVENT=DOWNLOAD_DONE") {
t.Errorf("Expected GOPEED_EVENT in output, got: %s", output)
}
+2 -7
View File
@@ -87,13 +87,8 @@ func TestScript_EnvironmentVariables(t *testing.T) {
t.Setenv("GOPEED_TEST_OUTPUT_FILE", outputFile)
downloader.triggerScripts(ScriptEventDownloadDone, task, nil)
waitForFile(t, outputFile, 5*time.Second)
content, err := os.ReadFile(outputFile)
if err != nil {
t.Fatalf("Failed to read output file: %v", err)
}
output := string(content)
// GOPEED_TASK_PATH is the last line the script writes.
output := waitForFileContains(t, outputFile, "GOPEED_TASK_PATH=", 5*time.Second)
if !strings.Contains(output, "GOPEED_EVENT=DOWNLOAD_DONE") {
t.Errorf("Expected GOPEED_EVENT in output, got: %s", output)
}
+28 -14
View File
@@ -190,14 +190,14 @@ func TestWebhook_NoWebhookConfigured(t *testing.T) {
}
func TestWebhook_MultipleUrls(t *testing.T) {
count := 0
received := make(chan string, 2)
server1 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
count++
received <- "server1"
w.WriteHeader(http.StatusOK)
}))
defer server1.Close()
server2 := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
count++
received <- "server2"
w.WriteHeader(http.StatusOK)
}))
defer server2.Close()
@@ -219,11 +219,21 @@ func TestWebhook_MultipleUrls(t *testing.T) {
// Trigger webhook
downloader.triggerWebhooks(WebhookEventDownloadDone, task, nil)
// Wait for webhooks
time.Sleep(500 * time.Millisecond)
if count != 2 {
t.Errorf("Expected 2 webhook calls, got %d", count)
// Wait for each destination, rather than racing on a shared counter or
// assuming that both requests finish within a fixed sleep.
timer := time.NewTimer(2 * time.Second)
defer timer.Stop()
seen := make(map[string]bool)
for len(seen) < 2 {
select {
case server := <-received:
if seen[server] {
t.Fatalf("Duplicate webhook call to %s", server)
}
seen[server] = true
case <-timer.C:
t.Fatalf("Timeout waiting for both webhook destinations; received: %v", seen)
}
}
})
}
@@ -629,9 +639,9 @@ func TestWebhook_SendWebhookToUrl_VariousStatusCodes(t *testing.T) {
}
func TestWebhook_TriggerWebhooks_EmptyUrlSkipped(t *testing.T) {
requestCount := 0
received := make(chan struct{}, 2)
server := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
requestCount++
received <- struct{}{}
w.WriteHeader(http.StatusOK)
}))
defer server.Close()
@@ -650,10 +660,14 @@ func TestWebhook_TriggerWebhooks_EmptyUrlSkipped(t *testing.T) {
downloader.triggerWebhooks(WebhookEventDownloadDone, task, nil)
time.Sleep(500 * time.Millisecond)
if requestCount != 2 {
t.Errorf("Expected 2 requests (empty URL should be skipped), got %d", requestCount)
timer := time.NewTimer(2 * time.Second)
defer timer.Stop()
for count := 0; count < 2; count++ {
select {
case <-received:
case <-timer.C:
t.Fatalf("Expected 2 requests (empty URL should be skipped), got %d", count)
}
}
})
}
@@ -53,6 +53,7 @@ class AppRuntimeState {
class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
static StartConfig? _defaultStartConfig;
Future<void> _preferenceWrites = Future<void>.value();
bool _analyticsStarted = false;
@override
Future<AppRuntimeState> build() async {
@@ -73,6 +74,7 @@ class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
startConfig
..apiEnable = true
..mcpEnable = config.api.mcpEnable;
await _initAnalytics(config);
} else {
startConfig
..apiEnable = config.api.enable
@@ -253,7 +255,8 @@ class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
}
Future<void> _initAnalytics(DownloaderConfig config) async {
if (!AnalyticsConfig.isConfigured || !config.extra.analyticsEnabled) return;
if (_analyticsStarted || !AnalyticsConfig.isConfigured || !config.extra.analyticsEnabled) return;
_analyticsStarted = true;
try {
final generatedClientId = await Analytics.instance.init(fallbackClientId: config.extra.deviceId);
if (generatedClientId != null) {
@@ -262,6 +265,7 @@ class AppRuntimeController extends AsyncNotifier<AppRuntimeState> {
}
unawaited(Analytics.instance.logAppOpen());
} catch (error, stackTrace) {
_analyticsStarted = false;
logger.w('GA4 init failed', error, stackTrace);
}
}
@@ -19,7 +19,7 @@ Future<DownloaderConfig> loadDownloaderConfig({
config = await load();
} catch (error, stackTrace) {
onLoadError?.call(error, stackTrace);
config = DownloaderConfig();
config = DownloaderConfig()..extra.analyticsEnabled = false;
loaded = false;
}
final changed = await initDownloaderConfig(config, defaultDownloadDir: defaultDownloadDir);
@@ -1,7 +1,9 @@
import 'package:flutter/foundation.dart' show TargetPlatform, defaultTargetPlatform;
import 'package:flutter/material.dart' show Colors;
import 'package:flutter/widgets.dart';
import 'package:window_manager/window_manager.dart';
import '../../shared/theme/app_palette.dart';
import '../../shared/widgets/window/desktop_window_header.dart';
import 'app_window_chrome.dart';
@@ -16,6 +18,15 @@ class AppWindowFrame extends StatelessWidget {
Widget build(BuildContext context) {
Widget framedChild = child;
// System insets apply to mobile devices at every responsive breakpoint.
// SafeArea also consumes the padding so page-level SafeAreas do not add it twice.
if (defaultTargetPlatform == TargetPlatform.android || defaultTargetPlatform == TargetPlatform.iOS) {
framedChild = ColoredBox(
color: AppPalette.of(context).bg,
child: SafeArea(child: framedChild),
);
}
if (AppWindowChrome.usesCustomChrome) {
framedChild = Stack(
children: [
@@ -19,6 +19,12 @@ class SettingsState {
}
}
class SettingsSaveResult {
const SettingsSaveResult({required this.btRestartRequired});
final bool btRestartRequired;
}
class SettingsController extends AsyncNotifier<SettingsState> {
Future<void> _saveTail = Future<void>.value();
int _saveRevision = 0;
@@ -36,7 +42,7 @@ class SettingsController extends AsyncNotifier<SettingsState> {
state = AsyncValue.data(SettingsState(config: config));
}
Future<void> save(DownloaderConfig config) {
Future<SettingsSaveResult> save(DownloaderConfig config) {
final snapshot = DownloaderConfig.fromJson(config.toJson());
final revision = ++_saveRevision;
state = AsyncValue.data(SettingsState(config: config, saving: true));
@@ -47,10 +53,15 @@ class SettingsController extends AsyncNotifier<SettingsState> {
return result;
}
Future<void> _save(DownloaderConfig config, int revision) async {
Future<SettingsSaveResult> _save(DownloaderConfig config, int revision) async {
try {
final service = ref.read(gopeedServiceProvider);
final latest = await service.getConfig();
final previousBt = (
latest.protocolConfig.bt.listenPort,
latest.protocolConfig.bt.disableDht,
latest.protocolConfig.bt.disablePex,
);
config.extra
..windowState = latest.extra.windowState
..bookmarks = latest.extra.bookmarks
@@ -64,6 +75,12 @@ class SettingsController extends AsyncNotifier<SettingsState> {
ref.read(appRuntimeControllerProvider.notifier).replaceDownloaderConfig(saved);
if (revision == _saveRevision) state = AsyncValue.data(SettingsState(config: saved));
}
final savedBt = (
saved.protocolConfig.bt.listenPort,
saved.protocolConfig.bt.disableDht,
saved.protocolConfig.bt.disablePex,
);
return SettingsSaveResult(btRestartRequired: previousBt != savedBt);
} finally {
if (ref.mounted && revision == _saveRevision && state.value?.saving == true) {
state = AsyncValue.data(state.requireValue.copyWith(saving: false));
@@ -625,7 +625,6 @@ class _SettingsPageState extends ConsumerState<SettingsPage> {
),
SettingsItem(
title: context.l10n.enableDht,
subtitle: context.l10n.restartRequired,
child: shad.Switch(
value: !config.protocolConfig.bt.disableDht,
onChanged: (value) => _mutateConfig((next) => next.protocolConfig.bt.disableDht = !value),
@@ -633,7 +632,6 @@ class _SettingsPageState extends ConsumerState<SettingsPage> {
),
SettingsItem(
title: context.l10n.enablePex,
subtitle: context.l10n.restartRequired,
child: shad.Switch(
value: !config.protocolConfig.bt.disablePex,
onChanged: (value) => _mutateConfig((next) => next.protocolConfig.bt.disablePex = !value),
@@ -1204,7 +1202,10 @@ class _SettingsPageState extends ConsumerState<SettingsPage> {
final snapshot = DownloaderConfig.fromJson(config.toJson());
_loadedSignature = _signature(snapshot);
try {
await ref.read(settingsControllerProvider.notifier).save(snapshot);
final result = await ref.read(settingsControllerProvider.notifier).save(snapshot);
if (mounted && result.btRestartRequired) {
_toast(context.l10n.restartRequired, type: AppToastType.info);
}
return true;
} catch (error) {
if (mounted) _toast(error.toString());
@@ -41,6 +41,7 @@ import '../../application/pending_download.dart';
import '../../application/task_list_navigation.dart';
import '../widgets/history_search_field.dart';
import '../widgets/create_task_action_buttons.dart';
import '../widgets/create_task_advanced_toggle.dart';
import '../widgets/resolve_download_dialog.dart';
class CreateTaskWindowPage extends ConsumerStatefulWidget {
@@ -560,20 +561,11 @@ class _CreateTaskWindowPageState extends ConsumerState<CreateTaskWindowPage> {
),
const SizedBox(height: 20),
Align(
alignment: Alignment.centerLeft,
child: GestureDetector(
onTap: _toggleAdvanced,
child: Row(
mainAxisSize: MainAxisSize.min,
children: [
Text(
context.l10n.advanced,
style: TextStyle(color: palette.textPrimary, fontSize: 14, fontWeight: FontWeight.w600),
),
const SizedBox(width: 8),
_TripleChevronIcon(expanded: _showAdvanced, color: palette.textSecondary),
],
),
alignment: AlignmentDirectional.centerStart,
child: CreateTaskAdvancedToggle(
key: const ValueKey('create-task-advanced-toggle'),
expanded: _showAdvanced,
onPressed: _toggleAdvanced,
),
),
AnimatedCrossFade(
@@ -1933,30 +1925,6 @@ class _MiniInputAction extends StatelessWidget {
}
}
class _TripleChevronIcon extends StatelessWidget {
const _TripleChevronIcon({required this.expanded, required this.color});
final bool expanded;
final Color color;
@override
Widget build(BuildContext context) {
final icon = expanded ? Icons.expand_less : Icons.expand_more;
return SizedBox(
width: 14,
height: 18,
child: Stack(
clipBehavior: Clip.none,
children: [
Positioned(top: 0, left: 0, child: Icon(icon, size: 14, color: color)),
Positioned(top: 4, left: 0, child: Icon(icon, size: 14, color: color)),
Positioned(top: 8, left: 0, child: Icon(icon, size: 14, color: color)),
],
),
);
}
}
class _SegmentButton extends StatelessWidget {
const _SegmentButton({required this.label, required this.active, required this.onTap});
@@ -0,0 +1,49 @@
import 'package:shadcn_flutter/shadcn_flutter.dart';
import '../../../../l10n/l10n.dart';
import '../../../../shared/theme/app_design_tokens.dart';
import '../../../../shared/theme/app_palette.dart';
/// A quiet disclosure control with the same keyboard behavior as other buttons.
class CreateTaskAdvancedToggle extends StatelessWidget {
const CreateTaskAdvancedToggle({super.key, required this.expanded, required this.onPressed});
final bool expanded;
final VoidCallback onPressed;
@override
Widget build(BuildContext context) {
final palette = AppPalette.of(context);
final reduceMotion = MediaQuery.disableAnimationsOf(context);
return Semantics(
expanded: expanded,
child: ButtonStyleOverride.inherit(
padding: (context, states, value) => const EdgeInsets.symmetric(vertical: AppDesignTokens.space12),
child: GhostButton(
onPressed: onPressed,
child: ConstrainedBox(
constraints: const BoxConstraints(minHeight: 20),
child: Row(
mainAxisSize: MainAxisSize.min,
children: [
Flexible(
child: Text(
context.l10n.advanced,
style: TextStyle(color: palette.textPrimary, fontSize: 14, fontWeight: FontWeight.w600),
),
),
const SizedBox(width: AppDesignTokens.space8),
AnimatedRotation(
turns: expanded ? 0.5 : 0,
duration: reduceMotion ? Duration.zero : const Duration(milliseconds: 180),
curve: Curves.easeOutCubic,
child: Icon(Icons.keyboard_arrow_down, size: 16, color: palette.textSecondary),
),
],
),
),
),
),
);
}
}
+1 -1
View File
@@ -208,7 +208,7 @@
"categoryName": "اسم الفئة",
"categoryPath": "مسار الفئة",
"autoCategorizeByExtension": "تصنيف تلقائي حسب نوع الملف",
"autoCategorizeByExtensionDescription": "توجيه التنزيلات إلى مجلد الفئة تلقائيًا حسب امتداد الملف عند الحفظ في مجلد التنزيل الافتراضي.",
"autoCategorizeByExtensionDescription": "توجيه التنزيلات إلى مجلد الفئة تلقائيًا حسب امتداد الملف عند الحفظ في مجلد التنزيل الافتراضي",
"categoryExtensions": "امتدادات الملفات",
"categoryExtensionsUsageHint": "افصل الامتدادات بفواصل أو مسافات ثم اضغط Enter.",
"categoryExtensionsEmptyHint": "بدون الامتدادات، لن تُحفظ الملفات في هذا التصنيف تلقائيًا.",
+1 -1
View File
@@ -215,7 +215,7 @@
"categoryName": "Kateqoriyanın Adı",
"categoryPath": "Kateqoriyanın Yolu",
"autoCategorizeByExtension": "Fayl tipinə görə avtomatik kateqoriyalaşdırma",
"autoCategorizeByExtensionDescription": "Standart yükləmə qovluğuna yadda saxlanarkən yükləmələr fayl uzantısına görə avtomatik olaraq kateqoriya qovluğuna yönləndirilir.",
"autoCategorizeByExtensionDescription": "Standart yükləmə qovluğuna yadda saxlanarkən yükləmələr fayl uzantısına görə avtomatik olaraq kateqoriya qovluğuna yönləndirilir",
"categoryExtensions": "Fayl uzantıları",
"categoryExtensionsUsageHint": "Uzantıları vergül və ya boşluqla ayırın, sonra Enter-a basın.",
"categoryExtensionsEmptyHint": "Uzantı yoxdursa, fayllar bu kateqoriyaya avtomatik olaraq saxlanmır.",
+1 -1
View File
@@ -208,7 +208,7 @@
"categoryName": "Nom de la categoria",
"categoryPath": "Ruta de la categoria",
"autoCategorizeByExtension": "Categoritza automàticament segons el tipus de fitxer",
"autoCategorizeByExtensionDescription": "Les baixades es desen a la carpeta de categoria segons l'extensió del fitxer en utilitzar la carpeta de baixades predeterminada.",
"autoCategorizeByExtensionDescription": "Les baixades es desen a la carpeta de categoria segons l'extensió del fitxer en utilitzar la carpeta de baixades predeterminada",
"categoryExtensions": "Extensions de fitxer",
"categoryExtensionsUsageHint": "Separa les extensions amb comes o espais i prem Retorn.",
"categoryExtensionsEmptyHint": "Sense extensions, els fitxers no es desaran aquí automàticament.",
Loaded 50 of 82 files, more files were not shown because too many files have changed in this diff. Show more