mirror of
https://github.com/GopeedLab/gopeed.git
synced 2026-11-10 10:30:28 +00:00
fix(extension): bound event execution and cancel pending engine calls
This commit is contained in:
1 parent
ed125c2f19
commit
1bc31774cb
7 files changed
+123
-17
No files matched your search
@@ -2405,7 +2405,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
|
||||
}
|
||||
|
||||
@@ -1,6 +1,8 @@
|
||||
package engine
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"github.com/dop251/goja"
|
||||
"testing"
|
||||
"time"
|
||||
@@ -32,3 +34,14 @@ func TestCloseUnblocksPendingPromise(t *testing.T) {
|
||||
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,22 @@ 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)
|
||||
e.cleanupMu.Lock()
|
||||
cleanups := e.cleanups
|
||||
e.cleanups = nil
|
||||
@@ -177,6 +210,7 @@ func NewEngine(cfg *Config) *Engine {
|
||||
loop := eventloop.NewEventLoop()
|
||||
engine := &Engine{
|
||||
loop: loop,
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
loop.Start()
|
||||
done := make(chan struct{})
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -122,3 +122,28 @@ func TestOnErrorCancellationReleasesTaskAPIs(t *testing.T) {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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
|
||||
|
||||
+12
-6
@@ -232,12 +232,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 +263,9 @@ func (cfg *DownloaderConfig) Init() *DownloaderConfig {
|
||||
new(ed2k.FetcherManager),
|
||||
}
|
||||
}
|
||||
if cfg.ExtensionEventTimeout <= 0 {
|
||||
cfg.ExtensionEventTimeout = time.Minute
|
||||
}
|
||||
if cfg.RefreshInterval == 0 {
|
||||
cfg.RefreshInterval = 350
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user