From 506f6899923ba2f284ede9f6b350c018b719b9c4 Mon Sep 17 00:00:00 2001 From: Levi Date: Sat, 14 Nov 2026 11:32:19 +0800 Subject: [PATCH] fix(extension): cancel stuck event handlers and unblock engine waiters (#1659) --- pkg/download/downloader.go | 4 +- pkg/download/engine/cancellation_test.go | 47 ++++++ pkg/download/engine/engine.go | 55 +++++++- pkg/download/extension.go | 19 ++- pkg/download/extension_cancellation_test.go | 149 ++++++++++++++++++++ pkg/download/extension_engine_session.go | 15 ++ pkg/download/model.go | 18 ++- 7 files changed, 290 insertions(+), 17 deletions(-) create mode 100644 pkg/download/engine/cancellation_test.go create mode 100644 pkg/download/extension_cancellation_test.go diff --git a/pkg/download/downloader.go b/pkg/download/downloader.go index 94ecbc4f..a76f3b45 100644 --- a/pkg/download/downloader.go +++ b/pkg/download/downloader.go @@ -2412,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 } diff --git a/pkg/download/engine/cancellation_test.go b/pkg/download/engine/cancellation_test.go new file mode 100644 index 00000000..d75b2ed7 --- /dev/null +++ b/pkg/download/engine/cancellation_test.go @@ -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) + } +} diff --git a/pkg/download/engine/engine.go b/pkg/download/engine/engine.go index 3dc20084..1deb92ca 100644 --- a/pkg/download/engine/engine.go +++ b/pkg/download/engine/engine.go @@ -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{}) diff --git a/pkg/download/extension.go b/pkg/download/extension.go index b160b0e4..94db2024 100644 --- a/pkg/download/extension.go +++ b/pkg/download/extension.go @@ -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 + } } } } diff --git a/pkg/download/extension_cancellation_test.go b/pkg/download/extension_cancellation_test.go new file mode 100644 index 00000000..362483cb --- /dev/null +++ b/pkg/download/extension_cancellation_test.go @@ -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) + } +} diff --git a/pkg/download/extension_engine_session.go b/pkg/download/extension_engine_session.go index ae617d7b..813aa3d6 100644 --- a/pkg/download/extension_engine_session.go +++ b/pkg/download/extension_engine_session.go @@ -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 diff --git a/pkg/download/model.go b/pkg/download/model.go index b08a34d3..60be919b 100644 --- a/pkg/download/model.go +++ b/pkg/download/model.go @@ -249,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 @@ -277,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 }