mirror of
https://github.com/GopeedLab/gopeed.git
synced 2026-11-15 12:53:41 +00:00
Compare commits
14
Commits
v2.0.0-beta.4
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
be0a395447 | ||
|
|
506f689992 | ||
|
|
38c2f1e8c2 | ||
|
|
0520fd2baf | ||
|
|
6f7fd056ef | ||
|
|
6cd95507fd | ||
|
|
89dfbe4c45 | ||
|
|
392ee36137 | ||
|
|
a34383a7e2 | ||
|
|
98879569f0 | ||
|
|
8ef3b63e8d | ||
|
|
d873cd411c | ||
|
|
9e9bcd4b01 | ||
|
|
5bd57608d2 |
No files matched your search
@@ -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 }}
|
||||
@@ -17,7 +17,7 @@ on:
|
||||
- "Dockerfile"
|
||||
|
||||
env:
|
||||
GO_VERSION: "1.25.4"
|
||||
GO_VERSION: "1.26.11"
|
||||
|
||||
permissions:
|
||||
contents: write
|
||||
|
||||
@@ -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
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"`
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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"
|
||||
)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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&lt;b.txt",
|
||||
want: "a<b.txt",
|
||||
},
|
||||
{
|
||||
name: "escaped ampersand before quote entity is decoded once",
|
||||
in: "a&quot;b.txt",
|
||||
want: "a"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)
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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 "&lt;" becomes "<" and is not decoded twice.
|
||||
var htmlEntityReplacer = strings.NewReplacer(
|
||||
"&", "&",
|
||||
"<", "<",
|
||||
">", ">",
|
||||
""", "\"",
|
||||
"'", "'",
|
||||
"'", "'",
|
||||
)
|
||||
|
||||
// unescapeHTMLEntities unescapes common HTML entities in filenames
|
||||
// This handles cases where servers HTML-encode special characters like & to &
|
||||
func unescapeHTMLEntities(s string) string {
|
||||
// Common HTML entities that might appear in filenames
|
||||
replacements := map[string]string{
|
||||
"&": "&",
|
||||
"<": "<",
|
||||
">": ">",
|
||||
""": "\"",
|
||||
"'": "'",
|
||||
"'": "'",
|
||||
}
|
||||
|
||||
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")
|
||||
}
|
||||
}
|
||||
@@ -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/",
|
||||
})
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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{})
|
||||
|
||||
@@ -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.
|
||||
@@ -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() {} }
|
||||
@@ -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}
|
||||
})
|
||||
}
|
||||
@@ -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) },
|
||||
})
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
@@ -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(),
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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),
|
||||
),
|
||||
],
|
||||
),
|
||||
),
|
||||
),
|
||||
),
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -208,7 +208,7 @@
|
||||
"categoryName": "اسم الفئة",
|
||||
"categoryPath": "مسار الفئة",
|
||||
"autoCategorizeByExtension": "تصنيف تلقائي حسب نوع الملف",
|
||||
"autoCategorizeByExtensionDescription": "توجيه التنزيلات إلى مجلد الفئة تلقائيًا حسب امتداد الملف عند الحفظ في مجلد التنزيل الافتراضي.",
|
||||
"autoCategorizeByExtensionDescription": "توجيه التنزيلات إلى مجلد الفئة تلقائيًا حسب امتداد الملف عند الحفظ في مجلد التنزيل الافتراضي",
|
||||
"categoryExtensions": "امتدادات الملفات",
|
||||
"categoryExtensionsUsageHint": "افصل الامتدادات بفواصل أو مسافات ثم اضغط Enter.",
|
||||
"categoryExtensionsEmptyHint": "بدون الامتدادات، لن تُحفظ الملفات في هذا التصنيف تلقائيًا.",
|
||||
|
||||
@@ -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.",
|
||||
|
||||
@@ -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
Reference in new issue
Block a user