mirror of
https://gh-proxy.org/https://github.com/klinsmaya/new-api.git
synced 2026-11-08 01:17:09 +08:00
utils.fetch is a synchronous, read-only GET or HEAD to the channel Base URL host or the plugin's allowedHosts, through the channel proxy used for polling. It is available in buildSubmitRequest, the query and batch build and parse hooks, and buildContentRequest; every other hook throws. Each hook call gets at most 4 fetches, 15 s per fetch, 30 s in total and 1 MiB per body. Redirects are returned, not followed. responseType "bytes" returns an ArrayBuffer for binary data such as a media file header. While a fetch waits, the runtime gives up its execution slot and the watchdog pauses. BuildContentRequest now receives the request context, and the media proxy shares the plugin request header rules.
899 lines
28 KiB
Go
899 lines
28 KiB
Go
package jsplugin
|
|
|
|
import (
|
|
"context"
|
|
"encoding"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"maps"
|
|
"math/big"
|
|
"reflect"
|
|
"regexp"
|
|
"slices"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
"unsafe"
|
|
|
|
"github.com/Calcium-Ion/moejs"
|
|
"github.com/QuantumNous/new-api/common"
|
|
"github.com/QuantumNous/new-api/logger"
|
|
)
|
|
|
|
const (
|
|
DefaultCallTimeout = 5 * time.Second
|
|
DefaultConcurrency = 8
|
|
)
|
|
|
|
var ErrCallAdmissionTimeout = errors.New("plugin call admission timed out")
|
|
|
|
const hookErrorMessageLimit = 512
|
|
|
|
// HookError reports a JavaScript exception thrown by a plugin hook. Message
|
|
// is the sanitized JS error message with engine prefixes stripped; it is safe
|
|
// to surface to API callers.
|
|
type HookError struct {
|
|
Hook string
|
|
Message string
|
|
wrapped error
|
|
}
|
|
|
|
func (e *HookError) Error() string {
|
|
if e == nil || e.wrapped == nil {
|
|
return "plugin hook failed"
|
|
}
|
|
return e.wrapped.Error()
|
|
}
|
|
|
|
func (e *HookError) Unwrap() error {
|
|
if e == nil {
|
|
return nil
|
|
}
|
|
return e.wrapped
|
|
}
|
|
|
|
// ResultError reports a hook result that CallInto could not store: one the
|
|
// host codec cannot encode (a NaN, an infinity, a cycle) or one the target's
|
|
// type rejects. Its message is the codec's.
|
|
type ResultError struct {
|
|
err error
|
|
}
|
|
|
|
func (e *ResultError) Error() string {
|
|
return e.err.Error()
|
|
}
|
|
|
|
func (e *ResultError) Unwrap() error {
|
|
return e.err
|
|
}
|
|
|
|
func newHookError(hook, rawMessage string, wrapped error) *HookError {
|
|
var b strings.Builder
|
|
b.Grow(len(rawMessage))
|
|
count := 0
|
|
for _, r := range rawMessage {
|
|
if count >= hookErrorMessageLimit {
|
|
break
|
|
}
|
|
if r < 0x20 || (r >= 0x80 && r <= 0x9F) {
|
|
r = ' '
|
|
}
|
|
b.WriteRune(r)
|
|
count++
|
|
}
|
|
message := b.String()
|
|
if message == "" {
|
|
message = "plugin hook failed"
|
|
}
|
|
return &HookError{Hook: hook, Message: message, wrapped: wrapped}
|
|
}
|
|
|
|
var forbiddenSyntax = regexp.MustCompile(`(?m)(^|[^A-Za-z0-9_$])(async|await|import)([^A-Za-z0-9_$]|$)`)
|
|
|
|
type Options struct {
|
|
Key string
|
|
Version string
|
|
Timeout time.Duration
|
|
Concurrency int
|
|
Now func() time.Time
|
|
Log func(string)
|
|
}
|
|
|
|
type Engine struct {
|
|
key string
|
|
version string
|
|
timeout time.Duration
|
|
now func() time.Time
|
|
log func(string)
|
|
module *moejs.Module
|
|
pool chan *runtimeInstance
|
|
semaphore chan struct{}
|
|
// parked bounds the runtimes that lent their slot to other calls while
|
|
// utils.fetch waits on the network; its capacity is the concurrency.
|
|
parked chan struct{}
|
|
hooksMu sync.RWMutex
|
|
hooks map[hookKey]moejs.Hook
|
|
// exports holds the exports that were not undefined once the module
|
|
// loaded; it is written only by Compile.
|
|
exports map[string]struct{}
|
|
}
|
|
|
|
// hookKey names a cached hook handle. Contract hooks are an export and at
|
|
// most two members (protocols.<name>.<hook>); longer paths only come from
|
|
// fixture requests and are resolved on every call.
|
|
type hookKey struct {
|
|
export string
|
|
members [2]string
|
|
depth int
|
|
}
|
|
|
|
// Fixture requests name arbitrary members, so the handle cache is bounded.
|
|
const maxCachedHooks = 256
|
|
|
|
type runtimeInstance struct {
|
|
runtime *moejs.Runtime
|
|
logContext *runtimeLogContext
|
|
// call is the hook call running on the runtime, nil between calls.
|
|
call *runtimeCall
|
|
}
|
|
|
|
type runtimeLogContext struct {
|
|
context context.Context
|
|
}
|
|
|
|
// Compile performs upload-time syntax checks and compiles an ESM plugin once.
|
|
// All engine-specific module and runtime handling is intentionally kept here.
|
|
func Compile(source string, options Options) (*Engine, error) {
|
|
if match := forbiddenSyntax.FindString(sourceWithoutCommentsAndStrings(source)); match != "" {
|
|
return nil, fmt.Errorf("unsupported plugin syntax %q: plugins must be synchronous and cannot import modules", strings.TrimSpace(match))
|
|
}
|
|
|
|
// The compiler never reads files, so a sourceMappingURL comment in
|
|
// untrusted source stays inert.
|
|
module, err := moejs.Compile(options.Key+".js", source)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("compile plugin: %w", err)
|
|
}
|
|
// export ... from has no import keyword; the host links no modules.
|
|
if requests := module.Requests(); len(requests) > 0 {
|
|
return nil, fmt.Errorf("unsupported plugin syntax: re-export from %q: plugins must be synchronous and cannot import modules", requests[0])
|
|
}
|
|
|
|
timeout := options.Timeout
|
|
if timeout <= 0 {
|
|
timeout = DefaultCallTimeout
|
|
}
|
|
concurrency := options.Concurrency
|
|
if concurrency <= 0 {
|
|
concurrency = DefaultConcurrency
|
|
}
|
|
now := options.Now
|
|
if now == nil {
|
|
now = time.Now
|
|
}
|
|
|
|
engine := &Engine{
|
|
key: options.Key,
|
|
version: options.Version,
|
|
timeout: timeout,
|
|
now: now,
|
|
log: options.Log,
|
|
module: module,
|
|
semaphore: make(chan struct{}, concurrency),
|
|
parked: make(chan struct{}, concurrency),
|
|
pool: make(chan *runtimeInstance, concurrency),
|
|
hooks: make(map[hookKey]moejs.Hook),
|
|
exports: make(map[string]struct{}),
|
|
}
|
|
instance, err := engine.newRuntime(context.Background())
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, name := range module.Exports() {
|
|
if value, found := instance.runtime.Export(name); found && !value.IsUndefined() {
|
|
engine.exports[name] = struct{}{}
|
|
}
|
|
}
|
|
instance.logContext.context = nil
|
|
engine.putRuntime(instance)
|
|
return engine, nil
|
|
}
|
|
|
|
// Export returns one module export without exposing engine values outside the
|
|
// engine boundary. It is used for declarative exports such as meta.
|
|
func (e *Engine) Export(ctx context.Context, exportName string) (any, error) {
|
|
select {
|
|
case e.semaphore <- struct{}{}:
|
|
defer func() { <-e.semaphore }()
|
|
case <-ctx.Done():
|
|
return nil, ctx.Err()
|
|
}
|
|
|
|
instance, err := e.getRuntime(ctx)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
reusable := true
|
|
defer func() {
|
|
instance.runtime.ClearInterrupt()
|
|
instance.logContext.context = nil
|
|
if reusable {
|
|
e.putRuntime(instance)
|
|
}
|
|
}()
|
|
timedOut := errors.New("plugin export timed out")
|
|
watchdog := watchRuntimeContext(instance.runtime, ctx, e.timeout, timedOut)
|
|
defer watchdog.stop()
|
|
value, found := instance.runtime.Export(exportName)
|
|
if !found || value.IsUndefined() {
|
|
return nil, fmt.Errorf("plugin export %q not found", exportName)
|
|
}
|
|
result, err := instance.runtime.ToGo(value)
|
|
if err == nil {
|
|
return result, nil
|
|
}
|
|
reusable = false
|
|
var interrupted *moejs.InterruptedError
|
|
if errors.As(err, &interrupted) {
|
|
return nil, fmt.Errorf("plugin %s@%s export %s interrupted: %w", e.key, e.version, exportName, interruptCause(interrupted))
|
|
}
|
|
var exception *moejs.Exception
|
|
errors.As(err, &exception)
|
|
return nil, fmt.Errorf("plugin %s@%s export %s failed: %w%s", e.key, e.version, exportName, err, thrownAt(instance.runtime, exception))
|
|
}
|
|
|
|
// HasExport reports whether a module export was defined when the plugin
|
|
// loaded: an export that was undefined then counts as absent, and a later
|
|
// assignment is not seen. Optional contract hooks should be detected with
|
|
// this method instead of relying on engine errors. It runs no JavaScript, so
|
|
// it neither waits for nor takes a call slot.
|
|
func (e *Engine) HasExport(exportName string) bool {
|
|
_, found := e.exports[exportName]
|
|
return found
|
|
}
|
|
|
|
// HasCallablePath reports whether an exported value, or a nested member below
|
|
// it, exists and is callable.
|
|
func (e *Engine) HasCallablePath(ctx context.Context, exportName string, members ...string) (bool, error) {
|
|
select {
|
|
case e.semaphore <- struct{}{}:
|
|
defer func() { <-e.semaphore }()
|
|
case <-ctx.Done():
|
|
return false, ctx.Err()
|
|
}
|
|
hook, err := e.hook(exportName, members)
|
|
if err != nil {
|
|
return false, nil
|
|
}
|
|
instance, err := e.getRuntime(ctx)
|
|
if err != nil {
|
|
return false, err
|
|
}
|
|
reusable := true
|
|
defer func() {
|
|
instance.runtime.ClearInterrupt()
|
|
instance.logContext.context = nil
|
|
if reusable {
|
|
e.putRuntime(instance)
|
|
}
|
|
}()
|
|
timedOut := errors.New("plugin inspection timed out")
|
|
watchdog := watchRuntimeContext(instance.runtime, ctx, e.timeout, timedOut)
|
|
defer watchdog.stop()
|
|
found, err := instance.runtime.Has(hook)
|
|
if err == nil {
|
|
return found, nil
|
|
}
|
|
reusable = false
|
|
var interrupted *moejs.InterruptedError
|
|
if errors.As(err, &interrupted) {
|
|
return false, fmt.Errorf("plugin %s@%s hook %s inspection interrupted: %w", e.key, e.version, hook.Name(), interruptCause(interrupted))
|
|
}
|
|
var exception *moejs.Exception
|
|
errors.As(err, &exception)
|
|
return false, fmt.Errorf("plugin %s@%s hook %s inspection failed: %w%s", e.key, e.version, hook.Name(), err, thrownAt(instance.runtime, exception))
|
|
}
|
|
|
|
// RawJSON is a hook argument given as JSON text: the hook receives JSON.parse
|
|
// of it, members in order, whether it is an argument or inside one, with no Go
|
|
// values built in between; text nested too deep for JSON.parse is decoded with
|
|
// the host codec instead. The engine parses the bytes in place, without the
|
|
// copy ParseJSON makes, and strings of the call's result may share them, so
|
|
// the caller must never change them once passed.
|
|
type RawJSON []byte
|
|
|
|
// Call invokes one named module export and returns its JSON-compatible value.
|
|
func (e *Engine) Call(ctx context.Context, exportName string, args ...any) (result any, err error) {
|
|
result, _, err = e.call(ctx, callOptions{}, exportName, nil, args...)
|
|
return result, err
|
|
}
|
|
|
|
// CallInto invokes one named module export and stores its result in target as
|
|
// the host codec decodes the encoding of the value Call returns, without that
|
|
// round trip for plain results. A result the codec cannot encode or target's
|
|
// type rejects is a *ResultError, and target may then hold part of it.
|
|
func (e *Engine) CallInto(ctx context.Context, target any, exportName string, args ...any) error {
|
|
_, _, err := e.call(ctx, callOptions{target: target}, exportName, nil, args...)
|
|
return err
|
|
}
|
|
|
|
// CallJSONInto is CallInto with the semantics of JSON.stringify's text of the
|
|
// result instead of the host codec's encoding: objects keep their member
|
|
// order, which a json.RawMessage in target receives as text, members whose
|
|
// value is undefined are left out and toJSON runs.
|
|
func (e *Engine) CallJSONInto(ctx context.Context, target any, exportName string, args ...any) error {
|
|
_, _, err := e.call(ctx, callOptions{target: target, stringify: true}, exportName, nil, args...)
|
|
return err
|
|
}
|
|
|
|
// CallMember invokes a function stored on an exported object, such as a
|
|
// renderer in the renderers export.
|
|
func (e *Engine) CallMember(ctx context.Context, exportName, memberName string, args ...any) (result any, err error) {
|
|
result, _, err = e.call(ctx, callOptions{}, exportName, []string{memberName}, args...)
|
|
return result, err
|
|
}
|
|
|
|
// CallPath invokes a function nested below an exported object. It is used for
|
|
// protocol hooks such as protocols.openai_responses.renderEvents.
|
|
func (e *Engine) CallPath(ctx context.Context, exportName string, members []string, args ...any) (result any, err error) {
|
|
result, _, err = e.call(ctx, callOptions{}, exportName, members, args...)
|
|
return result, err
|
|
}
|
|
|
|
// CallPathInto is CallInto for a function nested below an exported object.
|
|
func (e *Engine) CallPathInto(ctx context.Context, target any, exportName string, members []string, args ...any) error {
|
|
_, _, err := e.call(ctx, callOptions{target: target}, exportName, members, args...)
|
|
return err
|
|
}
|
|
|
|
// CallPathWithAdmissionTimeout gives long-lived observers a separate bound for
|
|
// waiting on JavaScript capacity. Once admitted, the hook receives the
|
|
// engine's full execution timeout instead of inheriting time already spent in
|
|
// the semaphore queue.
|
|
func (e *Engine) CallPathWithAdmissionTimeout(
|
|
ctx context.Context,
|
|
admissionTimeout time.Duration,
|
|
exportName string,
|
|
members []string,
|
|
args ...any,
|
|
) (result any, err error) {
|
|
result, _, err = e.call(ctx, callOptions{admissionTimeout: admissionTimeout}, exportName, members, args...)
|
|
return result, err
|
|
}
|
|
|
|
// CallPathWithMemberJSON is CallPathWithAdmissionTimeout that also returns the
|
|
// result's member as the text JSON.stringify writes for it, which keeps the
|
|
// member order of its objects. The text is nil when member is empty or the
|
|
// result has no such member.
|
|
func (e *Engine) CallPathWithMemberJSON(
|
|
ctx context.Context,
|
|
admissionTimeout time.Duration,
|
|
member string,
|
|
exportName string,
|
|
members []string,
|
|
args ...any,
|
|
) (result any, text json.RawMessage, err error) {
|
|
return e.call(ctx, callOptions{admissionTimeout: admissionTimeout, jsonMember: member}, exportName, members, args...)
|
|
}
|
|
|
|
// callOptions is what a hook call does besides calling: the bound on waiting
|
|
// for a JavaScript slot, the Go value the result is stored in instead of
|
|
// being returned (from its JSON.stringify text when stringify is set), and a
|
|
// member of the result also returned as JSON text.
|
|
type callOptions struct {
|
|
admissionTimeout time.Duration
|
|
target any
|
|
stringify bool
|
|
jsonMember string
|
|
}
|
|
|
|
func (e *Engine) call(ctx context.Context, opts callOptions, exportName string, members []string, args ...any) (any, json.RawMessage, error) {
|
|
if err := e.acquireCallSlot(ctx, opts.admissionTimeout); err != nil {
|
|
return nil, nil, err
|
|
}
|
|
call := &runtimeCall{engine: e, ctx: ctx, holdsSlot: true}
|
|
// utils.fetch may lend the slot out; return what the call holds now.
|
|
defer call.release()
|
|
|
|
hook, err := e.hook(exportName, members)
|
|
if err != nil {
|
|
if len(members) == 0 {
|
|
return nil, nil, fmt.Errorf("plugin export %q not found", exportName)
|
|
}
|
|
return nil, nil, fmt.Errorf("plugin hook %q not found", exportName)
|
|
}
|
|
instance, err := e.getRuntime(ctx)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
call.runtime, call.hook = instance.runtime, hook.Name()
|
|
call.fetcher, _ = ctx.Value(fetcherContextKey{}).(Fetcher)
|
|
instance.call = call
|
|
reusable := true
|
|
defer func() {
|
|
instance.runtime.ClearInterrupt()
|
|
instance.logContext.context = nil
|
|
instance.call = nil
|
|
if reusable {
|
|
// An idle pooled runtime must not keep this call's request
|
|
// data alive.
|
|
instance.runtime.ReleaseCallData()
|
|
e.putRuntime(instance)
|
|
}
|
|
}()
|
|
|
|
callArgs := make([]moejs.Value, 0, 4)
|
|
for index, arg := range args {
|
|
var converted moejs.Value
|
|
if raw, ok := arg.(RawJSON); !ok {
|
|
value, _ := pluginValue(instance.runtime, arg, 0)
|
|
converted, err = instance.runtime.FromGo(value)
|
|
} else if converted, err = instance.runtime.ParseJSONString(unsafe.String(unsafe.SliceData(raw), len(raw))); err != nil {
|
|
// JSON.parse stops at a lower nesting depth than the host codec.
|
|
var decoded any
|
|
if err = common.Unmarshal(raw, &decoded); err == nil {
|
|
converted, err = instance.runtime.FromGo(decoded)
|
|
}
|
|
}
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("plugin %s@%s hook %s argument %d: %w", e.key, e.version, hook.Name(), index+1, err)
|
|
}
|
|
callArgs = append(callArgs, converted)
|
|
}
|
|
|
|
timedOut := errors.New("plugin call timed out")
|
|
call.watchdog = watchRuntimeContext(instance.runtime, ctx, e.timeout, timedOut)
|
|
// utils.fetch pauses and restarts the watchdog; stop whichever watch is
|
|
// current when the call ends.
|
|
defer call.watchdog.stop()
|
|
|
|
value, err := instance.runtime.Call(hook, callArgs...)
|
|
var interrupted *moejs.InterruptedError
|
|
var exception *moejs.Exception
|
|
if err == nil {
|
|
var result any
|
|
var memberText json.RawMessage
|
|
switch {
|
|
case opts.target != nil && opts.stringify:
|
|
err = instance.runtime.Unmarshal(value, opts.target)
|
|
case opts.target != nil:
|
|
err = instance.runtime.ToGoInto(value, opts.target)
|
|
default:
|
|
result, err = instance.runtime.ToGo(value)
|
|
}
|
|
if err == nil && opts.jsonMember != "" {
|
|
// After the result's Go value: AppendJSON may run a toJSON or
|
|
// a getter, which ToGo never does.
|
|
var member moejs.Value
|
|
if member, err = instance.runtime.Get(value, opts.jsonMember); err == nil && !member.IsUndefined() {
|
|
memberText, err = instance.runtime.AppendJSON(nil, member)
|
|
}
|
|
}
|
|
if err == nil {
|
|
return result, memberText, nil
|
|
}
|
|
if errors.As(err, &interrupted) {
|
|
reusable = false
|
|
return nil, nil, fmt.Errorf("plugin %s@%s hook %s interrupted: %w", e.key, e.version, hook.Name(), interruptCause(interrupted))
|
|
}
|
|
// Past ToGo's errors, ToGoInto fails as json.Marshal and
|
|
// json.Unmarshal do, which is the host codec (common/json.go), and
|
|
// Unmarshal and AppendJSON as JSON.stringify does (a BigInt, a cycle).
|
|
var internal *moejs.InternalError
|
|
if (opts.target != nil || opts.jsonMember != "") && !errors.As(err, &exception) && !errors.As(err, &internal) {
|
|
return nil, nil, &ResultError{err: err}
|
|
}
|
|
}
|
|
switch {
|
|
case errors.Is(err, moejs.ErrHookNotFound):
|
|
if len(members) == 0 {
|
|
return nil, nil, fmt.Errorf("plugin export %q not found", exportName)
|
|
}
|
|
// Name the path up to the first missing member, as far as data
|
|
// properties show it; getters on the path are not run again.
|
|
hookName := exportName
|
|
current, found := instance.runtime.Export(exportName)
|
|
for _, member := range members {
|
|
if !found || current.IsNullish() {
|
|
break
|
|
}
|
|
hookName += "." + member
|
|
if !current.IsObject() {
|
|
break
|
|
}
|
|
current, found = current.AsObject().GetOwnDataValue(instance.runtime.Realm().KeyFromGoString(member))
|
|
}
|
|
return nil, nil, fmt.Errorf("plugin hook %q not found", hookName)
|
|
case errors.Is(err, moejs.ErrNotCallable):
|
|
return nil, nil, fmt.Errorf("plugin hook %q is not a function", hook.Name())
|
|
case errors.As(err, &interrupted):
|
|
reusable = false
|
|
return nil, nil, fmt.Errorf("plugin %s@%s hook %s failed: %w", e.key, e.version, hook.Name(), interruptCause(interrupted))
|
|
case errors.As(err, &exception):
|
|
wrapped := fmt.Errorf("plugin %s@%s hook %s failed: %w%s", e.key, e.version, hook.Name(), err, thrownAt(instance.runtime, exception))
|
|
return nil, nil, newHookError(hook.Name(), exception.Message(), wrapped)
|
|
}
|
|
reusable = false
|
|
return nil, nil, fmt.Errorf("plugin %s@%s hook %s failed: %w", e.key, e.version, hook.Name(), err)
|
|
}
|
|
|
|
// hook resolves a hook path against the module's export table. Only
|
|
// declared exports resolve; the handle is cached, while the bindings it
|
|
// names are read live on every call.
|
|
func (e *Engine) hook(exportName string, members []string) (moejs.Hook, error) {
|
|
key := hookKey{export: exportName, depth: len(members)}
|
|
cacheable := len(members) <= len(key.members)
|
|
if cacheable {
|
|
copy(key.members[:], members)
|
|
e.hooksMu.RLock()
|
|
hook, found := e.hooks[key]
|
|
e.hooksMu.RUnlock()
|
|
if found {
|
|
return hook, nil
|
|
}
|
|
}
|
|
hook, err := e.module.Hook(exportName, members...)
|
|
if err != nil || !cacheable {
|
|
return hook, err
|
|
}
|
|
e.hooksMu.Lock()
|
|
if len(e.hooks) < maxCachedHooks {
|
|
e.hooks[key] = hook
|
|
}
|
|
e.hooksMu.Unlock()
|
|
return hook, nil
|
|
}
|
|
|
|
func (e *Engine) acquireCallSlot(ctx context.Context, admissionTimeout time.Duration) error {
|
|
if admissionTimeout <= 0 {
|
|
select {
|
|
case e.semaphore <- struct{}{}:
|
|
return nil
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
}
|
|
}
|
|
|
|
timer := time.NewTimer(admissionTimeout)
|
|
defer timer.Stop()
|
|
select {
|
|
case e.semaphore <- struct{}{}:
|
|
return nil
|
|
case <-ctx.Done():
|
|
return ctx.Err()
|
|
case <-timer.C:
|
|
if err := ctx.Err(); err != nil {
|
|
return err
|
|
}
|
|
return fmt.Errorf("%w: plugin %s@%s", ErrCallAdmissionTimeout, e.key, e.version)
|
|
}
|
|
}
|
|
|
|
// interruptCause is what stopped a runtime: the context cause
|
|
// watchRuntimeContext interrupted it with, so errors.Is sees a caller's
|
|
// cancellation through the host error.
|
|
func interruptCause(interrupted *moejs.InterruptedError) error {
|
|
if cause, ok := interrupted.Value.(error); ok {
|
|
return cause
|
|
}
|
|
return interrupted
|
|
}
|
|
|
|
// thrownAt is the frame an Error was thrown from, as " at fn (file:line:col)",
|
|
// for host logs; it is empty for thrown values that are not Errors and for
|
|
// errors that are not exceptions (exception is nil).
|
|
func thrownAt(runtime *moejs.Runtime, exception *moejs.Exception) string {
|
|
_, frames, found := strings.Cut(runtime.StackTrace(exception), "\n at ")
|
|
if !found {
|
|
return ""
|
|
}
|
|
frame, _, _ := strings.Cut(frames, "\n")
|
|
return " at " + frame
|
|
}
|
|
|
|
// maxPluginValueDepth bounds the argument walk; deeper values, which no
|
|
// host-built argument reaches, are passed on as they are.
|
|
const maxPluginValueDepth = 512
|
|
|
|
// pluginValueContainers are the container types moejs converts; named types
|
|
// with one of these underlying types are only retyped.
|
|
var pluginValueContainers = []reflect.Type{
|
|
reflect.TypeFor[map[string]any](),
|
|
reflect.TypeFor[map[string]string](),
|
|
reflect.TypeFor[map[string][]string](),
|
|
reflect.TypeFor[[]any](),
|
|
reflect.TypeFor[[]string](),
|
|
reflect.TypeFor[[]map[string]any](),
|
|
}
|
|
|
|
// pluginValue returns a hook argument in the shapes moejs converts: nil,
|
|
// booleans, numbers, strings, JSON-shaped maps and slices and engine values.
|
|
// A RawJSON inside the argument becomes rt's parse of it (RawJSON); other Go
|
|
// values (structs, pointers, maps and slices of other element types) take
|
|
// their JSON form through the configured codec, and named scalars and
|
|
// containers become their underlying type. changed is false when v has
|
|
// supported shapes throughout, the usual case, so an argument is passed on
|
|
// without a copy.
|
|
func pluginValue(rt *moejs.Runtime, v any, depth int) (converted any, changed bool) {
|
|
if depth > maxPluginValueDepth {
|
|
return v, false
|
|
}
|
|
switch typed := v.(type) {
|
|
case nil, bool, string, int, int8, int16, int32, int64, uint, uint8, uint16, uint32, uint64,
|
|
float32, float64, json.Number, *big.Int, []byte, []string, map[string]string, map[string][]string,
|
|
moejs.Value, moejs.NativeFunc:
|
|
return v, false
|
|
case RawJSON:
|
|
if parsed, err := rt.ParseJSONString(unsafe.String(unsafe.SliceData(typed), len(typed))); err == nil {
|
|
return parsed, true
|
|
}
|
|
// JSON.parse stops at a lower nesting depth than the host codec.
|
|
var decoded any
|
|
if err := common.Unmarshal(typed, &decoded); err != nil {
|
|
return v, false
|
|
}
|
|
return decoded, true
|
|
case map[string]any:
|
|
var copied map[string]any
|
|
for key, item := range typed {
|
|
if item, changed = pluginValue(rt, item, depth+1); changed {
|
|
if copied == nil {
|
|
copied = maps.Clone(typed)
|
|
}
|
|
copied[key] = item
|
|
}
|
|
}
|
|
if copied == nil {
|
|
return v, false
|
|
}
|
|
return copied, true
|
|
case []any:
|
|
var copied []any
|
|
for index, item := range typed {
|
|
if item, changed = pluginValue(rt, item, depth+1); changed {
|
|
if copied == nil {
|
|
copied = slices.Clone(typed)
|
|
}
|
|
copied[index] = item
|
|
}
|
|
}
|
|
if copied == nil {
|
|
return v, false
|
|
}
|
|
return copied, true
|
|
case []map[string]any:
|
|
var copied []map[string]any
|
|
for index, item := range typed {
|
|
if next, changed := pluginValue(rt, item, depth+1); changed {
|
|
if copied == nil {
|
|
copied = slices.Clone(typed)
|
|
}
|
|
copied[index] = next.(map[string]any)
|
|
}
|
|
}
|
|
if copied == nil {
|
|
return v, false
|
|
}
|
|
return copied, true
|
|
}
|
|
_, marshaler := v.(json.Marshaler)
|
|
_, textMarshaler := v.(encoding.TextMarshaler)
|
|
if !marshaler && !textMarshaler {
|
|
value := reflect.ValueOf(v)
|
|
switch value.Kind() {
|
|
case reflect.Bool:
|
|
return value.Bool(), true
|
|
case reflect.Int, reflect.Int8, reflect.Int16, reflect.Int32, reflect.Int64:
|
|
return value.Int(), true
|
|
case reflect.Uint, reflect.Uint8, reflect.Uint16, reflect.Uint32, reflect.Uint64, reflect.Uintptr:
|
|
return value.Uint(), true
|
|
case reflect.Float32, reflect.Float64:
|
|
return value.Float(), true
|
|
case reflect.String:
|
|
return value.String(), true
|
|
case reflect.Map, reflect.Slice:
|
|
for _, container := range pluginValueContainers {
|
|
if value.Type().ConvertibleTo(container) {
|
|
converted, _ = pluginValue(rt, value.Convert(container).Interface(), depth)
|
|
return converted, true
|
|
}
|
|
}
|
|
}
|
|
}
|
|
data, err := common.Marshal(v)
|
|
if err != nil {
|
|
return v, false
|
|
}
|
|
var decoded any
|
|
if err = common.Unmarshal(data, &decoded); err != nil {
|
|
return v, false
|
|
}
|
|
return decoded, true
|
|
}
|
|
|
|
// Idle runtimes are bounded by the execution limit and survive GC. Start with
|
|
// one instance and grow only when concurrent work actually needs more.
|
|
func (e *Engine) putRuntime(instance *runtimeInstance) {
|
|
select {
|
|
case e.pool <- instance:
|
|
default:
|
|
}
|
|
}
|
|
|
|
func (e *Engine) getRuntime(ctx context.Context) (*runtimeInstance, error) {
|
|
select {
|
|
case instance := <-e.pool:
|
|
instance.logContext.context = ctx
|
|
return instance, nil
|
|
default:
|
|
return e.newRuntime(ctx)
|
|
}
|
|
}
|
|
|
|
// runtimeWatchdog interrupts a runtime with timeoutError once it has run for
|
|
// its timeout, or with the context's cause once ctx ends. utils.fetch pauses
|
|
// it, so the timeout counts JavaScript time only. A timeout callback must
|
|
// finish before its runtime can be reused: merely stopping a timer does not
|
|
// wait for an already-started Interrupt call.
|
|
type runtimeWatchdog struct {
|
|
runtime *moejs.Runtime
|
|
ctx context.Context
|
|
timeoutError error
|
|
remaining time.Duration
|
|
started time.Time
|
|
watch context.Context
|
|
stopWatch func() bool
|
|
cancel context.CancelFunc
|
|
interrupted chan struct{}
|
|
paused bool
|
|
}
|
|
|
|
func watchRuntimeContext(runtime *moejs.Runtime, ctx context.Context, timeout time.Duration, timeoutError error) *runtimeWatchdog {
|
|
watchdog := &runtimeWatchdog{runtime: runtime, ctx: ctx, timeoutError: timeoutError, remaining: timeout}
|
|
watchdog.start()
|
|
return watchdog
|
|
}
|
|
|
|
func (w *runtimeWatchdog) start() {
|
|
watch, cancel := context.WithTimeoutCause(w.ctx, w.remaining, w.timeoutError)
|
|
interrupted := make(chan struct{})
|
|
w.stopWatch = context.AfterFunc(watch, func() {
|
|
w.runtime.Interrupt(context.Cause(watch))
|
|
close(interrupted)
|
|
})
|
|
w.watch, w.cancel, w.interrupted = watch, cancel, interrupted
|
|
w.started, w.paused = time.Now(), false
|
|
}
|
|
|
|
func (w *runtimeWatchdog) stop() {
|
|
if w.paused {
|
|
return
|
|
}
|
|
if !w.stopWatch() {
|
|
<-w.interrupted
|
|
}
|
|
w.cancel()
|
|
}
|
|
|
|
// pause stops the clock while a host function waits. It reports false when
|
|
// the watchdog already fired: its interrupt is then ending the call, and the
|
|
// host function must return instead of waiting.
|
|
func (w *runtimeWatchdog) pause() bool {
|
|
if !w.stopWatch() {
|
|
<-w.interrupted
|
|
return false
|
|
}
|
|
w.cancel()
|
|
w.remaining -= time.Since(w.started)
|
|
w.paused = true
|
|
return true
|
|
}
|
|
|
|
// resume restarts the clock with the time left at pause and the same timeout
|
|
// cause. When the time is up or ctx ended meanwhile, it waits for the
|
|
// interrupt that follows, so the host function can return it.
|
|
func (w *runtimeWatchdog) resume() {
|
|
w.start()
|
|
if w.watch.Err() != nil {
|
|
<-w.interrupted
|
|
}
|
|
}
|
|
|
|
func (e *Engine) newRuntime(ctx context.Context) (*runtimeInstance, error) {
|
|
// Code compiled from strings would bypass the upload-time syntax checks,
|
|
// so eval and the Function constructors throw an EvalError.
|
|
runtime := moejs.NewRuntime(moejs.Options{DisableDynamicCode: true})
|
|
logContext := &runtimeLogContext{context: ctx}
|
|
instance := &runtimeInstance{runtime: runtime, logContext: logContext}
|
|
logOutput := e.log
|
|
if logOutput == nil {
|
|
logOutput = func(message string) {
|
|
logger.LogDebug(logContext.context, "task_plugin subsystem=runtime event=console message=%q", message)
|
|
}
|
|
}
|
|
if err := injectGlobals(runtime, func() string {
|
|
return fmt.Sprintf("[plugin:%s@%s]", e.key, e.version)
|
|
}, e.now, logOutput, func() *runtimeCall { return instance.call }); err != nil {
|
|
return nil, fmt.Errorf("inject plugin utils: %w", err)
|
|
}
|
|
timedOut := errors.New("plugin initialization timed out")
|
|
watchdog := watchRuntimeContext(runtime, ctx, e.timeout, timedOut)
|
|
err := runtime.Load(e.module)
|
|
watchdog.stop()
|
|
runtime.ClearInterrupt()
|
|
var interrupted *moejs.InterruptedError
|
|
if errors.As(err, &interrupted) {
|
|
return nil, fmt.Errorf("initialize plugin %s@%s: %w", e.key, e.version, interruptCause(interrupted))
|
|
}
|
|
if err != nil {
|
|
return nil, fmt.Errorf("evaluate plugin: %w", err)
|
|
}
|
|
return instance, nil
|
|
}
|
|
|
|
func sourceWithoutCommentsAndStrings(source string) string {
|
|
var output strings.Builder
|
|
output.Grow(len(source))
|
|
quote := byte(0)
|
|
escaped := false
|
|
lineComment := false
|
|
blockComment := false
|
|
for i := 0; i < len(source); i++ {
|
|
current := source[i]
|
|
next := byte(0)
|
|
if i+1 < len(source) {
|
|
next = source[i+1]
|
|
}
|
|
if lineComment {
|
|
if current == '\n' {
|
|
lineComment = false
|
|
output.WriteByte('\n')
|
|
} else {
|
|
output.WriteByte(' ')
|
|
}
|
|
continue
|
|
}
|
|
if blockComment {
|
|
if current == '*' && next == '/' {
|
|
blockComment = false
|
|
output.WriteString(" ")
|
|
i++
|
|
} else {
|
|
output.WriteByte(' ')
|
|
}
|
|
continue
|
|
}
|
|
if quote != 0 {
|
|
output.WriteByte(' ')
|
|
if escaped {
|
|
escaped = false
|
|
} else if current == '\\' {
|
|
escaped = true
|
|
} else if current == quote {
|
|
quote = 0
|
|
}
|
|
continue
|
|
}
|
|
if current == '/' && next == '/' {
|
|
lineComment = true
|
|
output.WriteString(" ")
|
|
i++
|
|
continue
|
|
}
|
|
if current == '/' && next == '*' {
|
|
blockComment = true
|
|
output.WriteString(" ")
|
|
i++
|
|
continue
|
|
}
|
|
if current == '\'' || current == '"' || current == '`' {
|
|
quote = current
|
|
output.WriteByte(' ')
|
|
continue
|
|
}
|
|
output.WriteByte(current)
|
|
}
|
|
return output.String()
|
|
}
|