Wip interrupt fix

js/eventloop/eventloop.go

@@ -6,6 +6,7 @@ import (
 	"sync"
 
 	"github.com/grafana/sobek"
+
 	"go.k6.io/k6/js/modules"
 )
 
@@ -79,6 +80,8 @@ func (e *EventLoop) wakeup() {
 // scenario has ended). Any error returned by any callback on the main thread
 // will abort the current iteration and no further event loop callbacks will be
 // executed in the same iteration.
+// As errors returned from resolve and reject are Interrupt errors those should
+// be returned.
 //
 // A common pattern for async work is something like this:
 //
@@ -94,11 +97,9 @@ func (e *EventLoop) wakeup() {
 //	        result, err := doTheActualAsyncWork(vu.Context())
 //	        enqueueCallback(func() error {
 //	            if err != nil {
-//	                reject(err)
-//	            } else {
-//	                resolve(result)
+//	                return reject(err)
 //	            }
-//	            return nil  // do not abort the iteration
+//	            return resolve(result)
 //	        })
 //	    }()
 //
@@ -124,7 +125,16 @@ func (e *EventLoop) RegisterCallback() (enqueueCallback func(func() error)) {
 			panic("RegisterCallback called twice")
 		}
 		callbackCalled = true
-		e.queue = append(e.queue, f)
+		wrapFunc := func() error {
+			rt := e.vu.Runtime()
+			callable, _ := sobek.AssertFunction(rt.ToValue(f))
+			_, err := callable(nil)
+			// if there is an exception the err will be nil either way
+
+			return err
+		}
+
+		e.queue = append(e.queue, wrapFunc)
 		e.registeredCallbacks--
 		e.lock.Unlock()
 		e.wakeup()

js/modules/cjsmodule.go

@@ -58,7 +58,7 @@ func (cmi *cjsModuleInstance) HasTLA() bool { return false }
 
 func (cmi *cjsModuleInstance) RequestedModules() []string { return cmi.w.RequestedModules() }
 
-func (cmi *cjsModuleInstance) ExecuteModule(rt *sobek.Runtime, _, _ func(any)) (sobek.CyclicModuleInstance, error) {
+func (cmi *cjsModuleInstance) ExecuteModule(rt *sobek.Runtime, _, _ func(any) error) (sobek.CyclicModuleInstance, error) {
 	v, err := rt.RunProgram(cmi.w.prg)
 	if err != nil {
 		return nil, err

js/modules/gomodule.go

@@ -60,7 +60,7 @@ type goModuleInstance struct {
 	defaultExport sobek.Value
 }
 
-func (gmi *goModuleInstance) ExecuteModule(_ *sobek.Runtime, _, _ func(any)) (sobek.CyclicModuleInstance, error) {
+func (gmi *goModuleInstance) ExecuteModule(_ *sobek.Runtime, _, _ func(any) error) (sobek.CyclicModuleInstance, error) {
 	return gmi, nil
 }
 func (gmi *goModuleInstance) HasTLA() bool { return false }

js/modules/gomodule_basic.go

@@ -67,6 +67,6 @@ func (bgmi *basicGoModuleInstance) GetBindingValue(n string) sobek.Value {
 
 func (bgmi *basicGoModuleInstance) HasTLA() bool { return false }
 
-func (bgmi *basicGoModuleInstance) ExecuteModule(_ *sobek.Runtime, _, _ func(any)) (sobek.CyclicModuleInstance, error) {
+func (bgmi *basicGoModuleInstance) ExecuteModule(_ *sobek.Runtime, _, _ func(any) error) (sobek.CyclicModuleInstance, error) {
 	return bgmi, nil
 }

js/modules/k6/experimental/fs/module.go

@@ -13,6 +13,7 @@ import (
 	"go.k6.io/k6/lib/fsext"
 
 	"github.com/grafana/sobek"
+
 	"go.k6.io/k6/js/common"
 	"go.k6.io/k6/js/modules"
 	"go.k6.io/k6/js/promises"
@@ -218,25 +219,25 @@ func (f *File) Stat() *sobek.Promise {
 //
 // It is possible for a read to successfully return with 0 bytes.
 // This does not indicate EOF.
-func (f *File) Read(into sobek.Value) *sobek.Promise {
+func (f *File) Read(into sobek.Value) (*sobek.Promise, error) {
 	promise, resolve, reject := f.vu.Runtime().NewPromise()
 
 	if common.IsNullish(into) {
-		reject(newFsError(TypeError, "read() failed; reason: into argument cannot be null or undefined"))
-		return promise
+		err := reject(newFsError(TypeError, "read() failed; reason: into argument cannot be null or undefined"))
+		return promise, err
 	}
 
 	intoObj := into.ToObject(f.vu.Runtime())
 	if !isUint8Array(f.vu.Runtime(), intoObj) {
-		reject(newFsError(TypeError, "read() failed; reason: into argument must be a Uint8Array"))
-		return promise
+		err := reject(newFsError(TypeError, "read() failed; reason: into argument must be a Uint8Array"))
+		return promise, err
 	}
 
 	// Obtain the underlying ArrayBuffer from the Uint8Array
 	ab, ok := intoObj.Get("buffer").Export().(sobek.ArrayBuffer)
 	if !ok {
-		reject(newFsError(TypeError, "read() failed; reason: into argument must be a Uint8Array"))
-		return promise
+		err := reject(newFsError(TypeError, "read() failed; reason: into argument must be a Uint8Array"))
+		return promise, err
 	}
 
 	// To avoid concurrency linked to modifying the runtime's `into` buffer from multiple
@@ -257,8 +258,7 @@ func (f *File) Read(into sobek.Value) *sobek.Promise {
 			// Read was successful, resolve early with the number of
 			// bytes read.
 			if readErr == nil {
-				resolve(n)
-				return nil
+				return resolve(n)
 			}
 
 			// If the read operation failed, we need to check if it was an io.EOF error
@@ -270,67 +270,61 @@ func (f *File) Read(into sobek.Value) *sobek.Promise {
 			var fsErr *fsError
 			isFSErr := errors.As(readErr, &fsErr)
 			if !isFSErr {
-				reject(readErr)
-				return nil
+				return reject(readErr)
 			}
 
 			if fsErr.kind == EOFError && n == 0 {
-				resolve(sobek.Null())
-			} else {
-				resolve(n)
+				return resolve(sobek.Null())
 			}
-
-			return nil
+			return resolve(n)
 		})
 	}()
 
-	return promise
+	return promise, nil
 }
 
 // Seek seeks to the given `offset` in the file, under the given `whence` mode.
 //
 // The returned promise resolves to the new `offset` (position) within the file, which
 // is expressed in bytes from the selected start, current, or end position depending
 // the provided `whence`.
-func (f *File) Seek(offset sobek.Value, whence sobek.Value) *sobek.Promise {
+func (f *File) Seek(offset sobek.Value, whence sobek.Value) (*sobek.Promise, error) {
 	promise, resolve, reject := f.vu.Runtime().NewPromise()
 
 	intOffset, err := exportInt(offset)
 	if err != nil {
-		reject(newFsError(TypeError, "seek() failed; reason: the offset argument "+err.Error()))
-		return promise
+		err := reject(newFsError(TypeError, "seek() failed; reason: the offset argument "+err.Error()))
+		return promise, err
 	}
 
 	intWhence, err := exportInt(whence)
 	if err != nil {
-		reject(newFsError(TypeError, "seek() failed; reason: the whence argument "+err.Error()))
-		return promise
+		err := reject(newFsError(TypeError, "seek() failed; reason: the whence argument "+err.Error()))
+		return promise, err
 	}
 
 	seekMode := SeekMode(intWhence)
 	switch seekMode {
 	case SeekModeStart, SeekModeCurrent, SeekModeEnd:
 		// Valid modes, do nothing.
 	default:
-		reject(newFsError(TypeError, "seek() failed; reason: the whence argument must be a SeekMode"))
-		return promise
+		err := reject(newFsError(TypeError, "seek() failed; reason: the whence argument must be a SeekMode"))
+		return promise, err
 	}
 
 	callback := f.vu.RegisterCallback()
 	go func() {
 		newOffset, err := f.ReadSeekStater.Seek(intOffset, seekMode)
 		callback(func() error {
 			if err != nil {
-				reject(err)
-				return err
+				return reject(err)
 			}
 
-			resolve(newOffset)
-			return nil
+			return resolve(newOffset)
 		})
 	}()
 
-	return promise
+	return promise, nil
 }
 
 func isUint8Array(rt *sobek.Runtime, o *sobek.Object) bool {

js/modules/k6/experimental/streams/readable_stream_default_reader.go

@@ -2,6 +2,7 @@
 
 import (
 	"github.com/grafana/sobek"
+
 	"go.k6.io/k6/js/common"
 )
 
@@ -63,15 +64,15 @@
 	readRequest := ReadRequest{
 		chunkSteps: func(chunk any) {
 			// Resolve promise with «[ "value" → chunk, "done" → false ]».
 			resolve(map[string]any{"value": chunk, "done": false})
 		},
 		closeSteps: func() {
 			// Resolve promise with «[ "value" → undefined, "done" → true ]».
 			resolve(map[string]any{"value": sobek.Undefined(), "done": true})
 		},
 		errorSteps: func(e any) {
 			// Reject promise with e.
 			reject(e)
 		},
 	}
 

js/modules/k6/experimental/streams/readable_stream_reader.go

@@ -2,6 +2,7 @@
 
 import (
 	"github.com/grafana/sobek"
+
 	"go.k6.io/k6/js/common"
 	"go.k6.io/k6/js/modules"
 )
@@ -35,10 +36,10 @@
 	SetStream(stream *ReadableStream)
 
 	// GetClosed returns a [sobek.Promise] that resolves when the stream is closed.
-	GetClosed() (p *sobek.Promise, resolve func(any), reject func(any))
+	GetClosed() (p *sobek.Promise, resolve, reject func(any) error)
 
 	// SetClosed sets the [sobek.Promise] that resolves when the stream is closed.
-	SetClosed(p *sobek.Promise, resolve func(any), reject func(any))
+	SetClosed(p *sobek.Promise, resolve, reject func(any) error)
 
 	// Cancel returns a [sobek.Promise] that resolves when the stream is canceled.
 	Cancel(reason sobek.Value) *sobek.Promise
@@ -47,8 +48,8 @@
 // BaseReadableStreamReader is a base implement
 type BaseReadableStreamReader struct {
 	closedPromise            *sobek.Promise
-	closedPromiseResolveFunc func(resolve any)
-	closedPromiseRejectFunc  func(reason any)
+	closedPromiseResolveFunc func(resolve any) error
+	closedPromiseRejectFunc  func(reason any) error
 
 	// stream is a [ReadableStream] instance that owns this reader
 	stream *ReadableStream
@@ -73,12 +74,12 @@
 }
 
 // GetClosed returns the reader's closed promise as well as its resolve and reject functions.
-func (reader *BaseReadableStreamReader) GetClosed() (p *sobek.Promise, resolve func(any), reject func(any)) {
+func (reader *BaseReadableStreamReader) GetClosed() (p *sobek.Promise, resolve, reject func(any) error) {
 	return reader.closedPromise, reader.closedPromiseResolveFunc, reader.closedPromiseRejectFunc
 }
 
 // SetClosed sets the reader's closed promise as well as its resolve and reject functions.
-func (reader *BaseReadableStreamReader) SetClosed(p *sobek.Promise, resolve func(any), reject func(any)) {
+func (reader *BaseReadableStreamReader) SetClosed(p *sobek.Promise, resolve, reject func(any) error) {
 	reader.closedPromise = p
 	reader.closedPromiseResolveFunc = resolve
 	reader.closedPromiseRejectFunc = reject
@@ -133,7 +134,7 @@
 
 	// 4. If stream.[[state]] is "readable", reject reader.[[closedPromise]] with a TypeError exception.
 	if stream.state == ReadableStreamStateReadable {
 		reader.closedPromiseRejectFunc(newTypeError(reader.runtime, "stream is readable").Err())
 	} else { // 5. Otherwise, set reader.[[closedPromise]] to a promise rejected with a TypeError exception.
 		reader.closedPromise = newRejectedPromise(stream.vu, newTypeError(reader.runtime, "stream is not readable").Err())
 	}
@@ -196,7 +197,7 @@
 	// 4. Otherwise, if stream.[[state]] is "closed",
 	case ReadableStreamStateClosed:
 		// 4.1 Set reader.[[closedPromise]] to a promise resolved with undefined.
 		resolve(sobek.Undefined())
 	// 5. Otherwise,
 	default:
 		// 5.1 Assert: stream.[[state]] is "errored".
@@ -206,9 +207,9 @@
 
 		// 5.2 Set reader.[[closedPromise]] to a promise rejected with stream.[[storedError]].
 		if jsErr, ok := stream.storedError.(*jsError); ok {
 			reject(jsErr.Err())
 		} else {
 			reject(errToObj(stream.runtime, stream.storedError))
 		}
 
 		// 5.3 Set reader.[[closedPromise]].[[PromiseIsHandled]] to true.

js/modules/k6/grpc/client.go

@@ -292,13 +292,13 @@ func (c *Client) AsyncInvoke(
 	method string,
 	req sobek.Value,
 	params sobek.Value,
-) *sobek.Promise {
+) (*sobek.Promise, error) {
 	grpcReq, err := c.buildInvokeRequest(method, req, params)
 
 	promise, resolve, reject := c.vu.Runtime().NewPromise()
 	if err != nil {
-		reject(err)
-		return promise
+		err = reject(err)
+		return promise, err
 	}
 
 	callback := c.vu.RegisterCallback()
@@ -307,15 +307,13 @@ func (c *Client) AsyncInvoke(
 
 		callback(func() error {
 			if err != nil {
-				reject(err)
-				return nil //nolint:nilerr // we don't want to return the error
+				return reject(err)
 			}
-			resolve(res)
-			return nil
+			return resolve(res)
 		})
 	}()
 
-	return promise
+	return promise, nil
 }
 
 // buildInvokeRequest creates a new InvokeRequest from the given method name, request object and parameters

js/modules/k6/http/request.go

@@ -97,11 +97,11 @@ func (c *Client) asyncRequest(method string, url sobek.Value, args ...sobek.Valu
 	if err != nil {
 		var resp *Response
 		if resp, err = c.handleParseRequestError(err); err != nil {
-			reject(err)
+			err = reject(err)
 		} else {
-			resolve(resp)
+			err = resolve(resp)
 		}
-		return p, nil
+		return p, err
 	}
 
 	callback := c.moduleInstance.vu.RegisterCallback()
@@ -110,12 +110,10 @@ func (c *Client) asyncRequest(method string, url sobek.Value, args ...sobek.Valu
 		resp, err := httpext.MakeRequest(c.moduleInstance.vu.Context(), state, req)
 		callback(func() error {
 			if err != nil {
-				reject(err)
-				return nil //nolint:nilerr // we want to reject the promise in this case
+				return reject(err)
 			}
 			c.processResponse(resp, req.ResponseType)
-			resolve(c.responseFromHTTPext(resp))
-			return nil
+			return resolve(c.responseFromHTTPext(resp))
 		})
 	}()
 

js/promises/promises.go

@@ -3,6 +3,7 @@ package promises
 
 import (
 	"github.com/grafana/sobek"
+
 	"go.k6.io/k6/js/modules"
 )
 
@@ -33,15 +34,13 @@ func New(vu modules.VU) (p *sobek.Promise, resolve func(result any), reject func
 
 	resolve = func(result any) {
 		callback(func() error {
-			resolveFunc(result)
-			return nil
+			return resolveFunc(result)
 		})
 	}
 
 	reject = func(reason any) {
 		callback(func() error {
-			rejectFunc(reason)
-			return nil
+			return rejectFunc(reason)
 		})
 	}