fix: prevent bus congestion by flushing out messages

cmd/cli/command/app/run.go

@@ -250,10 +250,10 @@ type ModuleDepFunc func(*moduleDeps) error
 
 func getServerModules(deps *moduleDeps) []app.ServerModuleFactory {
 	return []app.ServerModuleFactory{
+		module.LifecycleModuleFactory(),
 		module.ContextModuleFactory(),
 		module.ConsoleModuleFactory(),
 		cast.CastModuleFactory(),
-		module.LifecycleModuleFactory(),
 		netModule.ModuleFactory(deps.Bus),
 		module.RPCModuleFactory(deps.Bus),
 		module.StoreModuleFactory(deps.DocumentStore),

pkg/app/promise_proxy.go

@@ -39,3 +39,14 @@ func NewPromiseProxy(promise *goja.Promise, resolve func(result interface{}), re
 
 	return proxy
 }
+
+func NewPromiseProxyFrom(rt *goja.Runtime) *PromiseProxy {
+	promise, resolve, reject := rt.NewPromise()
+
+	return NewPromiseProxy(promise, resolve, reject)
+}
+
+func IsPromise(v goja.Value) (*goja.Promise, bool) {
+	promise, ok := v.Export().(*goja.Promise)
+	return promise, ok
+}

pkg/app/server.go

@@ -17,15 +17,22 @@ var (
 )
 
 type Server struct {
-	runtime *goja.Runtime
 	loop      *eventloop.EventLoop
+	factories []ServerModuleFactory
 	modules   []ServerModule
 }
 
 func (s *Server) Load(name string, src string) error {
-	_, err := s.runtime.RunScript(name, src)
+	var err error
+
+	s.loop.RunOnLoop(func(rt *goja.Runtime) {
+		_, err = rt.RunScript(name, src)
 		if err != nil {
-		return errors.Wrap(err, "could not run js script")
+			err = errors.Wrap(err, "could not run js script")
+		}
+	})
+	if err != nil {
+		return errors.WithStack(err)
 	}
 
 	return nil
@@ -34,15 +41,15 @@ func (s *Server) Load(name string, src string) error {
 func (s *Server) ExecFuncByName(ctx context.Context, funcName string, args ...interface{}) (goja.Value, error) {
 	ctx = logger.With(ctx, logger.F("function", funcName), logger.F("args", args))
 
-	callable, ok := goja.AssertFunction(s.runtime.Get(funcName))
+	ret, err := s.Exec(ctx, funcName, args...)
-	if !ok {
+	if err != nil {
-		return nil, errors.WithStack(ErrFuncDoesNotExist)
+		return nil, errors.WithStack(err)
 	}
 
-	return s.Exec(ctx, callable, args...)
+	return ret, nil
 }
 
-func (s *Server) Exec(ctx context.Context, callable goja.Callable, args ...interface{}) (goja.Value, error) {
+func (s *Server) Exec(ctx context.Context, callableOrFuncname any, args ...interface{}) (goja.Value, error) {
 	var (
 		wg    sync.WaitGroup
 		value goja.Value
@@ -51,7 +58,28 @@ func (s *Server) Exec(ctx context.Context, callable goja.Callable, args ...inter
 
 	wg.Add(1)
 
-	s.loop.RunOnLoop(func(vm *goja.Runtime) {
+	s.loop.RunOnLoop(func(rt *goja.Runtime) {
+		var callable goja.Callable
+		switch typ := callableOrFuncname.(type) {
+		case goja.Callable:
+			callable = typ
+
+		case string:
+			call, ok := goja.AssertFunction(rt.Get(typ))
+			if !ok {
+				err = errors.WithStack(ErrFuncDoesNotExist)
+
+				return
+			}
+
+			callable = call
+
+		default:
+			err = errors.Errorf("callableOrFuncname: expected callable or function name, got '%T'", callableOrFuncname)
+
+			return
+		}
+
 		logger.Debug(ctx, "executing callable")
 
 		defer wg.Done()
@@ -73,7 +101,7 @@ func (s *Server) Exec(ctx context.Context, callable goja.Callable, args ...inter
 
 		jsArgs := make([]goja.Value, 0, len(args))
 		for _, a := range args {
-			jsArgs = append(jsArgs, vm.ToValue(a))
+			jsArgs = append(jsArgs, rt.ToValue(a))
 		}
 
 		value, err = callable(nil, jsArgs...)
@@ -84,12 +112,11 @@ func (s *Server) Exec(ctx context.Context, callable goja.Callable, args ...inter
 
 	wg.Wait()
 
-	return value, err
+	if err != nil {
+		return nil, errors.WithStack(err)
 	}
 
-func (s *Server) IsPromise(v goja.Value) (*goja.Promise, bool) {
+	return value, nil
-	promise, ok := v.Export().(*goja.Promise)
-	return promise, ok
 }
 
 func (s *Server) WaitForPromise(promise *goja.Promise) goja.Value {
@@ -135,28 +162,21 @@ func (s *Server) WaitForPromise(promise *goja.Promise) goja.Value {
 	return value
 }
 
-func (s *Server) NewPromise() *PromiseProxy {
-	promise, resolve, reject := s.runtime.NewPromise()
-
-	return NewPromiseProxy(promise, resolve, reject)
-}
-
-func (s *Server) ToValue(v interface{}) goja.Value {
-	return s.runtime.ToValue(v)
-}
-
 func (s *Server) Start() error {
 	s.loop.Start()
 
-	for _, mod := range s.modules {
+	var err error
-		initMod, ok := mod.(InitializableModule)
-		if !ok {
-			continue
-		}
 
-		if err := initMod.OnInit(); err != nil {
+	s.loop.RunOnLoop(func(rt *goja.Runtime) {
-			return errors.WithStack(err)
+		rt.SetFieldNameMapper(goja.TagFieldNameMapper("goja", true))
+		rt.SetRandSource(createRandomSource())
+
+		if err = s.initModules(rt); err != nil {
+			err = errors.WithStack(err)
 		}
+	})
+	if err != nil {
+		return errors.WithStack(err)
 	}
 
 	return nil
@@ -166,36 +186,46 @@ func (s *Server) Stop() {
 	s.loop.Stop()
 }
 
-func (s *Server) initModules(factories ...ServerModuleFactory) {
+func (s *Server) initModules(rt *goja.Runtime) error {
-	runtime := goja.New()
+	modules := make([]ServerModule, 0, len(s.factories))
 
-	runtime.SetFieldNameMapper(goja.TagFieldNameMapper("goja", true))
+	for _, moduleFactory := range s.factories {
-	runtime.SetRandSource(createRandomSource())
-
-	modules := make([]ServerModule, 0, len(factories))
-
-	for _, moduleFactory := range factories {
 		mod := moduleFactory(s)
-		export := runtime.NewObject()
+
+		export := rt.NewObject()
 		mod.Export(export)
-		runtime.Set(mod.Name(), export)
+
+		rt.Set(mod.Name(), export)
 
 		modules = append(modules, mod)
 	}
 
-	s.runtime = runtime
+	for _, mod := range modules {
+		initMod, ok := mod.(InitializableModule)
+		if !ok {
+			continue
+		}
+
+		logger.Debug(context.Background(), "initializing module", logger.F("module", initMod.Name()))
+
+		if err := initMod.OnInit(rt); err != nil {
+			return errors.WithStack(err)
+		}
+	}
+
 	s.modules = modules
+
+	return nil
 }
 
 func NewServer(factories ...ServerModuleFactory) *Server {
 	server := &Server{
+		factories: factories,
 		loop: eventloop.NewEventLoop(
 			eventloop.EnableConsole(false),
 		),
 	}
 
-	server.initModules(factories...)
-
 	return server
 }
 

pkg/app/server_module.go

@@ -13,5 +13,5 @@ type ServerModule interface {
 
 type InitializableModule interface {
 	ServerModule
-	OnInit() error
+	OnInit(rt *goja.Runtime) error
 }

pkg/bus/memory/bus.go

@@ -22,13 +22,13 @@ func (b *Bus) Subscribe(ctx context.Context, ns bus.MessageNamespace) (<-chan bu
 	)
 
 	dispatchers := b.getDispatchers(ns)
-	d := newEventDispatcher(b.opt.BufferSize)
+	disp := newEventDispatcher(b.opt.BufferSize)
 
-	go d.Run()
+	go disp.Run(ctx)
 
-	dispatchers.Add(d)
+	dispatchers.Add(disp)
 
-	return d.Out(), nil
+	return disp.Out(), nil
 }
 
 func (b *Bus) Unsubscribe(ctx context.Context, ns bus.MessageNamespace, ch <-chan bus.Message) {
@@ -52,6 +52,12 @@ func (b *Bus) Publish(ctx context.Context, msg bus.Message) error {
 	)
 
 	for _, d := range dispatchersList {
+		if d.Closed() {
+			dispatchers.Remove(d)
+
+			continue
+		}
+
 		if err := d.In(msg); err != nil {
 			return errors.WithStack(err)
 		}

pkg/bus/memory/event_dispatcher.go

@@ -1,9 +1,13 @@
 package memory
 
 import (
+	"context"
 	"sync"
+	"time"
 
 	"forge.cadoles.com/arcad/edge/pkg/bus"
+	"github.com/pkg/errors"
+	"gitlab.com/wpetit/goweb/logger"
 )
 
 type eventDispatcherSet struct {
@@ -18,13 +22,21 @@ func (s *eventDispatcherSet) Add(d *eventDispatcher) {
 	s.items[d] = struct{}{}
 }
 
+func (s *eventDispatcherSet) Remove(d *eventDispatcher) {
+	s.mutex.Lock()
+	defer s.mutex.Unlock()
+
+	d.close()
+	delete(s.items, d)
+}
+
 func (s *eventDispatcherSet) RemoveByOutChannel(out <-chan bus.Message) {
 	s.mutex.Lock()
 	defer s.mutex.Unlock()
 
 	for d := range s.items {
 		if d.IsOut(out) {
-			d.Close()
+			d.close()
 			delete(s.items, d)
 		}
 	}
@@ -56,10 +68,21 @@ type eventDispatcher struct {
 	closed bool
 }
 
+func (d *eventDispatcher) Closed() bool {
+	d.mutex.RLock()
+	defer d.mutex.RUnlock()
+
+	return d.closed
+}
+
 func (d *eventDispatcher) Close() {
 	d.mutex.Lock()
 	defer d.mutex.Unlock()
 
+	d.close()
+}
+
+func (d *eventDispatcher) close() {
 	d.closed = true
 	close(d.in)
 }
@@ -85,16 +108,52 @@ func (d *eventDispatcher) IsOut(out <-chan bus.Message) bool {
 	return d.out == out
 }
 
-func (d *eventDispatcher) Run() {
+func (d *eventDispatcher) Run(ctx context.Context) {
+	defer func() {
+		for {
+			logger.Debug(ctx, "closing dispatcher, flushing out incoming messages")
+
+			close(d.out)
+
+			// Flush all incoming messages
+			for {
+				_, ok := <-d.in
+				if !ok {
+					return
+				}
+			}
+		}
+	}()
+
 	for {
 		msg, ok := <-d.in
 		if !ok {
-			close(d.out)
-
 			return
 		}
 
-		d.out <- msg
+		timeout := time.After(time.Second)
+
+		select {
+		case d.out <- msg:
+		case <-timeout:
+			logger.Error(
+				ctx,
+				"out message channel timeout",
+				logger.F("message", msg),
+			)
+
+			return
+
+		case <-ctx.Done():
+			logger.Error(
+				ctx,
+				"message subscription context canceled",
+				logger.F("message", msg),
+				logger.E(errors.WithStack(ctx.Err())),
+			)
+
+			return
+		}
 	}
 }
 

pkg/module/app/mount.go

@@ -73,12 +73,7 @@ func (h *Handler) serveAppURL(w http.ResponseWriter, r *http.Request) {
 
 	from := req.From
 	if from == "" {
-		host, _, err := net.SplitHostPort(r.RemoteAddr)
+		from = retrieveRemoteAddr(r)
-		if err != nil {
-			logger.Warn(ctx, "could not split remote address", logger.E(errors.WithStack(err)))
-		} else {
-			from = host
-		}
 	}
 
 	url, err := h.repo.GetURL(ctx, appID, from)
@@ -110,3 +105,12 @@ func Mount(repository Repository) MountFunc {
 		r.Post("/api/v1/apps/{appID}/url", handler.serveAppURL)
 	}
 }
+
+func retrieveRemoteAddr(r *http.Request) string {
+	host, _, err := net.SplitHostPort(r.RemoteAddr)
+	if err != nil {
+		host = r.RemoteAddr
+	}
+
+	return host
+}

pkg/module/auth/module.go

@@ -53,6 +53,10 @@ func (m *Module) Export(export *goja.Object) {
 	if err := export.Set("CLAIM_PREFERRED_USERNAME", ClaimPreferredUsername); err != nil {
 		panic(errors.Wrap(err, "could not set 'CLAIM_PREFERRED_USERNAME' property"))
 	}
+
+	if err := export.Set("CLAIM_ISSUER", ClaimIssuer); err != nil {
+		panic(errors.Wrap(err, "could not set 'CLAIM_ISSUER' property"))
+	}
 }
 
 func (m *Module) getClaim(call goja.FunctionCall, rt *goja.Runtime) goja.Value {

pkg/module/cast/module.go

@@ -15,10 +15,7 @@ const (
 	defaultTimeout = 30 * time.Second
 )
 
-type Module struct {
+type Module struct{}
-	ctx    context.Context
-	server *app.Server
-}
 
 func (m *Module) Name() string {
 	return "cast"
@@ -54,7 +51,7 @@ func (m *Module) refreshDevices(call goja.FunctionCall, rt *goja.Runtime) goja.V
 		panic(rt.ToValue(errors.WithStack(err)))
 	}
 
-	promise := m.server.NewPromise()
+	promise := app.NewPromiseProxyFrom(rt)
 
 	go func() {
 		ctx, cancel := context.WithTimeout(context.Background(), timeout)
@@ -102,7 +99,7 @@ func (m *Module) loadUrl(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
 		panic(rt.ToValue(errors.WithStack(err)))
 	}
 
-	promise := m.server.NewPromise()
+	promise := app.NewPromiseProxyFrom(rt)
 
 	go func() {
 		ctx, cancel := context.WithTimeout(context.Background(), timeout)
@@ -121,7 +118,7 @@ func (m *Module) loadUrl(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
 		promise.Resolve(nil)
 	}()
 
-	return m.server.ToValue(promise)
+	return rt.ToValue(promise)
 }
 
 func (m *Module) stopCast(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
@@ -137,7 +134,7 @@ func (m *Module) stopCast(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
 		panic(rt.ToValue(errors.WithStack(err)))
 	}
 
-	promise := m.server.NewPromise()
+	promise := app.NewPromiseProxyFrom(rt)
 
 	go func() {
 		ctx, cancel := context.WithTimeout(context.Background(), timeout)
@@ -156,7 +153,7 @@ func (m *Module) stopCast(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
 		promise.Resolve(nil)
 	}()
 
-	return m.server.ToValue(promise)
+	return rt.ToValue(promise)
 }
 
 func (m *Module) getStatus(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
@@ -172,7 +169,7 @@ func (m *Module) getStatus(call goja.FunctionCall, rt *goja.Runtime) goja.Value
 		panic(rt.ToValue(errors.WithStack(err)))
 	}
 
-	promise := m.server.NewPromise()
+	promise := app.NewPromiseProxyFrom(rt)
 
 	go func() {
 		ctx, cancel := context.WithTimeout(context.Background(), timeout)
@@ -191,7 +188,7 @@ func (m *Module) getStatus(call goja.FunctionCall, rt *goja.Runtime) goja.Value
 		promise.Resolve(status)
 	}()
 
-	return m.server.ToValue(promise)
+	return rt.ToValue(promise)
 }
 
 func (m *Module) parseTimeout(rawTimeout string) (time.Duration, error) {
@@ -214,8 +211,6 @@ func (m *Module) parseTimeout(rawTimeout string) (time.Duration, error) {
 
 func CastModuleFactory() app.ServerModuleFactory {
 	return func(server *app.Server) app.ServerModule {
-		return &Module{
+		return &Module{}
-			server: server,
-		}
 	}
 }

pkg/module/cast/module_test.go

@@ -85,7 +85,7 @@ func TestCastModuleRefreshDevices(t *testing.T) {
 		t.Error(errors.WithStack(err))
 	}
 
-	promise, ok := server.IsPromise(result)
+	promise, ok := app.IsPromise(result)
 	if !ok {
 		t.Fatal("expected promise")
 	}

pkg/module/lifecycle.go

@@ -9,9 +9,7 @@ import (
 	"gitlab.com/wpetit/goweb/logger"
 )
 
-type LifecycleModule struct {
+type LifecycleModule struct{}
-	server *app.Server
-}
 
 func (m *LifecycleModule) Name() string {
 	return "lifecycle"
@@ -20,25 +18,37 @@ func (m *LifecycleModule) Name() string {
 func (m *LifecycleModule) Export(export *goja.Object) {
 }
 
-func (m *LifecycleModule) OnInit() error {
+func (m *LifecycleModule) OnInit(rt *goja.Runtime) (err error) {
-	if _, err := m.server.ExecFuncByName(context.Background(), "onInit"); err != nil {
+	call, ok := goja.AssertFunction(rt.Get("onInit"))
-		if errors.Is(err, app.ErrFuncDoesNotExist) {
+	if !ok {
-			logger.Warn(context.Background(), "could not find onInit() function", logger.E(errors.WithStack(err)))
+		logger.Warn(context.Background(), "could not find onInit() function")
 
 		return nil
 	}
 
-		return errors.WithStack(err)
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			revoveredErr, ok := recovered.(error)
+			if ok {
+				logger.Error(context.Background(), "recovered runtime error", logger.E(errors.WithStack(revoveredErr)))
+
+				err = errors.WithStack(app.ErUnknownError)
+
+				return
 			}
 
+			panic(recovered)
+		}
+	}()
+
+	call(nil)
+
 	return nil
 }
 
 func LifecycleModuleFactory() app.ServerModuleFactory {
 	return func(server *app.Server) app.ServerModule {
-		module := &LifecycleModule{
+		module := &LifecycleModule{}
-			server: server,
-		}
 
 		return module
 	}

pkg/module/rpc.go

@@ -51,6 +51,12 @@ func (m *RPCModule) Export(export *goja.Object) {
 	}
 }
 
+func (m *RPCModule) OnInit(rt *goja.Runtime) error {
+	go m.handleMessages()
+
+	return nil
+}
+
 func (m *RPCModule) register(call goja.FunctionCall, rt *goja.Runtime) goja.Value {
 	fnName := util.AssertString(call.Argument(0), rt)
 
@@ -117,16 +123,21 @@ func (m *RPCModule) handleMessages() {
 	}
 
 	for msg := range clientMessages {
+		go m.handleMessage(ctx, msg, sendRes)
+	}
+}
+
+func (m *RPCModule) handleMessage(ctx context.Context, msg bus.Message, sendRes func(ctx context.Context, req *RPCRequest, result goja.Value)) {
 	clientMessage, ok := msg.(*ClientMessage)
 	if !ok {
 		logger.Warn(ctx, "unexpected bus message", logger.F("message", msg))
 
-			continue
+		return
 	}
 
 	ok, req := m.isRPCRequest(clientMessage)
 	if !ok {
-			continue
+		return
 	}
 
 	logger.Debug(ctx, "received rpc request", logger.F("request", req))
@@ -143,7 +154,7 @@ func (m *RPCModule) handleMessages() {
 			)
 		}
 
-			continue
+		return
 	}
 
 	callable, ok := rawCallable.(goja.Callable)
@@ -158,7 +169,7 @@ func (m *RPCModule) handleMessages() {
 			)
 		}
 
-			continue
+		return
 	}
 
 	result, err := m.server.Exec(clientMessage.Context, callable, clientMessage.Context, req.Params)
@@ -178,10 +189,10 @@ func (m *RPCModule) handleMessages() {
 			)
 		}
 
-			continue
+		return
 	}
 
-		promise, ok := m.server.IsPromise(result)
+	promise, ok := app.IsPromise(result)
 	if ok {
 		go func(ctx context.Context, req *RPCRequest, promise *goja.Promise) {
 			result := m.server.WaitForPromise(promise)
@@ -191,7 +202,6 @@ func (m *RPCModule) handleMessages() {
 		sendRes(clientMessage.Context, req, result)
 	}
 }
-}
 
 func (m *RPCModule) sendErrorResponse(ctx context.Context, req *RPCRequest, err error) error {
 	return m.sendResponse(ctx, &RPCResponse{
@@ -263,8 +273,8 @@ func RPCModuleFactory(bus bus.Bus) app.ServerModuleFactory {
 			bus:    bus,
 		}
 
-		go mod.handleMessages()
-
 		return mod
 	}
 }
+
+var _ app.InitializableModule = &RPCModule{}