perf: share websocket streams, cache settings reads, and parallelize auto-heal inspections

Author: kmendell

backend/internal/api/ws_handler.go

@@ -0,0 +1,163 @@
+package api
+
+import (
+	"context"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	systemtypes "github.com/getarcaneapp/arcane/types/system"
+	"github.com/gin-gonic/gin"
+	"github.com/gorilla/websocket"
+	"github.com/stretchr/testify/require"
+)
+
+func newTestWebSocketHandler() *WebSocketHandler {
+	return &WebSocketHandler{
+		wsMetrics:  NewWebSocketMetrics(),
+		logStreams: make(map[string]*wsLogStream),
+		wsUpgrader: websocket.Upgrader{
+			CheckOrigin: func(r *http.Request) bool { return true },
+		},
+	}
+}
+
+func dialWebSocket(t *testing.T, serverURL, path string) *websocket.Conn {
+	t.Helper()
+
+	wsURL := "ws" + strings.TrimPrefix(serverURL, "http") + path
+	conn, resp, err := websocket.DefaultDialer.Dial(wsURL, nil)
+	require.NoError(t, err)
+	if resp != nil {
+		_ = resp.Body.Close()
+	}
+
+	return conn
+}
+
+func TestWebSocketHandler_ProjectLogs_SharedStreamPerTarget(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	handler := newTestWebSocketHandler()
+	var starts atomic.Int32
+	var cancels atomic.Int32
+
+	handler.projectLogStreamer = func(ctx context.Context, projectID string, logsChan chan<- string, follow bool, tail, since string, timestamps bool) error {
+		starts.Add(1)
+		ticker := time.NewTicker(20 * time.Millisecond)
+		defer ticker.Stop()
+		defer cancels.Add(1)
+
+		for {
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			case <-ticker.C:
+				select {
+				case <-ctx.Done():
+					return ctx.Err()
+				case logsChan <- "api | shared project log":
+				}
+			}
+		}
+	}
+
+	router := gin.New()
+	router.GET("/api/environments/:id/ws/projects/:projectId/logs", handler.ProjectLogs)
+	server := httptest.NewServer(router)
+	defer server.Close()
+
+	conn1 := dialWebSocket(t, server.URL, "/api/environments/0/ws/projects/project-1/logs")
+	conn2 := dialWebSocket(t, server.URL, "/api/environments/0/ws/projects/project-1/logs")
+
+	_ = conn1.SetReadDeadline(time.Now().Add(2 * time.Second))
+	_, _, err := conn1.ReadMessage()
+	require.NoError(t, err)
+
+	_ = conn2.SetReadDeadline(time.Now().Add(2 * time.Second))
+	_, _, err = conn2.ReadMessage()
+	require.NoError(t, err)
+
+	require.Eventually(t, func() bool {
+		return starts.Load() == 1
+	}, 2*time.Second, 20*time.Millisecond)
+
+	require.Eventually(t, func() bool {
+		handler.logStreamsMu.Lock()
+		defer handler.logStreamsMu.Unlock()
+		return len(handler.logStreams) == 1
+	}, time.Second, 20*time.Millisecond)
+
+	require.NoError(t, conn1.Close())
+
+	time.Sleep(100 * time.Millisecond)
+
+	handler.logStreamsMu.Lock()
+	activeAfterFirstClose := len(handler.logStreams)
+	handler.logStreamsMu.Unlock()
+	require.Equal(t, 1, activeAfterFirstClose)
+	require.Equal(t, int32(0), cancels.Load())
+
+	require.NoError(t, conn2.Close())
+
+	require.Eventually(t, func() bool {
+		handler.logStreamsMu.Lock()
+		defer handler.logStreamsMu.Unlock()
+		return len(handler.logStreams) == 0
+	}, 2*time.Second, 20*time.Millisecond)
+	require.Eventually(t, func() bool {
+		return cancels.Load() == 1
+	}, 2*time.Second, 20*time.Millisecond)
+}
+
+func TestWebSocketHandler_SystemStats_UsesSharedSampler(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+
+	handler := newTestWebSocketHandler()
+	var collects atomic.Int32
+
+	handler.systemStatsCollector = func(ctx context.Context) systemtypes.SystemStats {
+		n := collects.Add(1)
+		return systemtypes.SystemStats{
+			CPUUsage: float64(n),
+		}
+	}
+
+	router := gin.New()
+	router.GET("/api/environments/:id/ws/system/stats", handler.SystemStats)
+	server := httptest.NewServer(router)
+	defer server.Close()
+
+	conn1 := dialWebSocket(t, server.URL, "/api/environments/0/ws/system/stats?interval=1")
+	conn2 := dialWebSocket(t, server.URL, "/api/environments/0/ws/system/stats?interval=1")
+
+	_ = conn1.SetReadDeadline(time.Now().Add(2 * time.Second))
+	_, _, err := conn1.ReadMessage()
+	require.NoError(t, err)
+
+	_ = conn2.SetReadDeadline(time.Now().Add(2 * time.Second))
+	_, _, err = conn2.ReadMessage()
+	require.NoError(t, err)
+
+	time.Sleep(1200 * time.Millisecond)
+
+	afterWarmup := collects.Load()
+	require.GreaterOrEqual(t, afterWarmup, int32(1))
+	require.LessOrEqual(t, afterWarmup, int32(3))
+
+	require.NoError(t, conn1.Close())
+	require.NoError(t, conn2.Close())
+
+	require.Eventually(t, func() bool {
+		handler.systemStatsSampler.lifecycleMu.Lock()
+		defer handler.systemStatsSampler.lifecycleMu.Unlock()
+		return !handler.systemStatsSampler.running && handler.systemStatsSampler.clients == 0
+	}, 2*time.Second, 20*time.Millisecond)
+
+	stoppedAt := collects.Load()
+	time.Sleep(1200 * time.Millisecond)
+	require.Equal(t, stoppedAt, collects.Load())
+}

backend/internal/api/ws_handler_test.go

@@ -5,9 +5,9 @@ import (
 	"errors"
 	"fmt"
 	"reflect"
-	"slices"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -21,6 +21,21 @@ type SettingVariable struct {
 	Value string
 }
 
+type settingFieldMeta struct {
+	index       int
+	key         string
+	attrs       string
+	isPublic    bool
+	isSensitive bool
+	isLocal     bool
+}
+
+var settingsFieldCache struct {
+	once    sync.Once
+	ordered []settingFieldMeta
+	byKey   map[string]settingFieldMeta
+}
+
 // IsTrue returns true if the value is a truthy string
 func (s SettingVariable) IsTrue() bool {
 	ok, _ := strconv.ParseBool(s.Value)
@@ -150,29 +165,63 @@ func (SettingVariable) TableName() string {
 	return "settings"
 }
 
-func (s *Settings) ToSettingVariableSlice(showAll bool, redactSensitiveValues bool) []SettingVariable {
-	cfgValue := reflect.ValueOf(s).Elem()
-	cfgType := cfgValue.Type()
-
-	var res []SettingVariable
-
-	for i := 0; i < cfgType.NumField(); i++ {
-		field := cfgType.Field(i)
+func buildSettingsFieldCache() {
+	rt := reflect.TypeFor[Settings]()
+	ordered := make([]settingFieldMeta, 0, rt.NumField())
+	byKey := make(map[string]settingFieldMeta, rt.NumField())
 
+	for i := 0; i < rt.NumField(); i++ {
+		field := rt.Field(i)
 		key, attrs, _ := strings.Cut(field.Tag.Get("key"), ",")
 		if key == "" {
 			continue
 		}
 
-		if !showAll && !strings.Contains(attrs, "public") {
+		meta := settingFieldMeta{
+			index:       i,
+			key:         key,
+			attrs:       attrs,
+			isPublic:    strings.Contains(attrs, "public"),
+			isSensitive: strings.Contains(attrs, "sensitive"),
+			isLocal:     strings.Contains(attrs, "local"),
+		}
+		ordered = append(ordered, meta)
+		byKey[key] = meta
+	}
+
+	settingsFieldCache.ordered = ordered
+	settingsFieldCache.byKey = byKey
+}
+
+func getSettingsFieldCache() ([]settingFieldMeta, map[string]settingFieldMeta) {
+	settingsFieldCache.once.Do(buildSettingsFieldCache)
+	return settingsFieldCache.ordered, settingsFieldCache.byKey
+}
+
+func (s *Settings) Clone() *Settings {
+	if s == nil {
+		return &Settings{}
+	}
+
+	clone := *s
+	return &clone
+}
+
+func (s *Settings) ToSettingVariableSlice(showAll bool, redactSensitiveValues bool) []SettingVariable {
+	cfgValue := reflect.ValueOf(s).Elem()
+	fields, _ := getSettingsFieldCache()
+
+	res := make([]SettingVariable, 0, len(fields))
+	for _, field := range fields {
+		if !showAll && !field.isPublic {
 			continue
 		}
 
-		value := cfgValue.Field(i).FieldByName("Value").String()
-		value = redactSettingValue(key, value, attrs, redactSensitiveValues)
+		value := cfgValue.Field(field.index).FieldByName("Value").String()
+		value = redactSettingValue(field.key, value, field.attrs, redactSensitiveValues)
 
 		settingVariable := SettingVariable{
-			Key:   key,
+			Key:   field.key,
 			Value: value,
 		}
 		res = append(res, settingVariable)
@@ -183,65 +232,47 @@ func (s *Settings) ToSettingVariableSlice(showAll bool, redactSensitiveValues bo
 
 func (s *Settings) FieldByKey(key string) (defaultValue string, isPublic bool, isSensitive bool, err error) {
 	rv := reflect.ValueOf(s).Elem()
-	rt := rv.Type()
-
-	for i := 0; i < rt.NumField(); i++ {
-		tagValue := strings.Split(rt.Field(i).Tag.Get("key"), ",")
-		keyFromTag := tagValue[0]
-		isPublic = slices.Contains(tagValue, "public")
-		isSensitive = slices.Contains(tagValue, "sensitive")
+	_, byKey := getSettingsFieldCache()
 
-		if keyFromTag != key {
-			continue
-		}
-
-		valueField := rv.Field(i).FieldByName("Value")
-		return valueField.String(), isPublic, isSensitive, nil
+	field, ok := byKey[key]
+	if !ok {
+		return "", false, false, SettingKeyNotFoundError{field: key}
 	}
 
-	return "", false, false, SettingKeyNotFoundError{field: key}
+	valueField := rv.Field(field.index).FieldByName("Value")
+	return valueField.String(), field.isPublic, field.isSensitive, nil
 }
 
 func (s *Settings) IsLocalSetting(key string) bool {
-	rt := reflect.TypeFor[Settings]()
-
-	for field := range rt.Fields() {
-		tagValue := strings.Split(field.Tag.Get("key"), ",")
-		keyFromTag := tagValue[0]
-
-		if keyFromTag == key {
-			return slices.Contains(tagValue, "local")
-		}
+	_, byKey := getSettingsFieldCache()
+	field, ok := byKey[key]
+	if !ok {
+		return false
 	}
 
-	return false
+	return field.isLocal
 }
 
 func (s *Settings) UpdateField(key string, value string, noSensitive bool) error {
 	rv := reflect.ValueOf(s).Elem()
-	rt := rv.Type()
-
-	for i := 0; i < rt.NumField(); i++ {
-		tagValue, attrs, _ := strings.Cut(rt.Field(i).Tag.Get("key"), ",")
-		if tagValue != key {
-			continue
-		}
+	_, byKey := getSettingsFieldCache()
 
-		// If the field is sensitive and noSensitive is true, we skip that
-		if noSensitive && strings.Contains(attrs, "sensitive") {
-			return SettingSensitiveForbiddenError{field: key}
-		}
+	field, ok := byKey[key]
+	if !ok {
+		return SettingKeyNotFoundError{field: key}
+	}
 
-		valueField := rv.Field(i).FieldByName("Value")
-		if !valueField.CanSet() {
-			return fmt.Errorf("field Value in SettingVariable is not settable for config key '%s'", key)
-		}
+	if noSensitive && field.isSensitive {
+		return SettingSensitiveForbiddenError{field: key}
+	}
 
-		valueField.SetString(value)
-		return nil
+	valueField := rv.Field(field.index).FieldByName("Value")
+	if !valueField.CanSet() {
+		return fmt.Errorf("field Value in SettingVariable is not settable for config key '%s'", key)
 	}
 
-	return SettingKeyNotFoundError{field: key}
+	valueField.SetString(value)
+	return nil
 }
 
 // helper keeps redaction logic in one place; behavior unchanged