fix(websocket): improve write and read deadlines (#1693)

Author: StarpTech

.github/workflows/router-ci.yaml

@@ -136,8 +136,13 @@ jobs:
     runs-on: ubuntu-latest
     timeout-minutes: 30
     services:
+      nats:
+        image: ghcr.io/wundergraph/cosmo/nats:2.11.0
+        ports:
+          - 4222:4222
+          - 8222:8222
+          - 6222:6222
       redis:
-        # Docker Hub image
         image: redis:7
         # Set health checks to wait until redis has started
         options: >-

docker-compose.yml

@@ -202,13 +202,14 @@ services:
       - dev
 
   nats:
-    image: nats:${DC_NATS_VERSION:-2.10.6}
+    image: ghcr.io/wundergraph/cosmo/nats:${DC_NATS_VERSION:-2.11.0}
+    build:
+      context: .
+      dockerfile: docker/nats/Dockerfile
     ports:
       - '4222:4222'
       - '8222:8222'
       - '6222:6222'
-    command:
-      - '--js'
     networks:
       - primary
     profiles:

docker/nats/Dockerfile

@@ -0,0 +1,10 @@
+FROM nats:2.11.0
+
+# Copy custom config
+COPY nats-server.conf /etc/nats/nats-server.conf
+
+# Expose the default ports
+EXPOSE 4222 8222 6222
+
+# Run NATS with JetStream using custom config
+CMD ["-c", "/etc/nats/nats-server.conf"]

docker/nats/README.md

@@ -0,0 +1,13 @@
+# Nats
+
+This image is only used for development and testing purposes. It is not intended for production use.
+
+__Reason:__ We haven't found any docker image that can run NATS with JetStream enabled. This is needed for GitHub Actions. We tried [Bitnami's image](https://hub.docker.com/r/bitnami/nats) and Nats official but it doesn't work.
+
+## Build & Release
+
+Run the following command to build and push the image for `linux/amd64`,`linux/arm64` and push it to the registry:
+
+```bash
+./build-push.sh
+```

docker/nats/build-push.sh

@@ -0,0 +1,23 @@
+#!/bin/bash
+
+# Exit on any error
+set -e
+
+# --- Config ---
+REPOSITORY="wundergraph/cosmo"
+IMAGE_NAME="nats"
+IMAGE_TAG="2.11.0"
+GHCR_IMAGE="ghcr.io/${REPOSITORY}/${IMAGE_NAME}:${IMAGE_TAG}"
+
+# Create and use a buildx builder (if not exists)
+docker buildx create --name multi-builder --use --bootstrap || true
+
+# Build the Docker image
+# Build and push multi-arch image
+echo "🔨 Building multi-arch image for linux/amd64 and linux/arm64..."
+docker buildx build \
+  --platform linux/amd64,linux/arm64 \
+  -t "$GHCR_IMAGE" \
+  --push .
+
+echo "✅ Done! Multi-arch image pushed to $GHCR_IMAGE"

docker/nats/nats-server.conf

@@ -0,0 +1,7 @@
+jetstream {
+  store_dir: /data/jetstream
+  max_mem_store: 1Gb
+  max_file_store: 10Gb
+}
+
+server_name: "mynats"

router-tests/Makefile

@@ -11,6 +11,9 @@ test-fresh: clean-testcache test
 test-no-race:
 	go test ./...
 
+test-no-race-fresh: clean-testcache
+	go test ./...
+
 lint:
 	go vet ./...
 	staticcheck ./...

router-tests/cache_warmup_test.go

@@ -6,13 +6,13 @@ import (
 	"testing"
 	"time"
 
+	nodev1 "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/node/v1"
 	"go.opentelemetry.io/otel/attribute"
 	"go.opentelemetry.io/otel/sdk/metric"
-	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 	"go.opentelemetry.io/otel/sdk/metric/metricdata/metricdatatest"
 
-	nodev1 "github.com/wundergraph/cosmo/router/gen/proto/wg/cosmo/node/v1"
 	"github.com/wundergraph/cosmo/router/pkg/otel"
+	"go.opentelemetry.io/otel/sdk/metric/metricdata"
 
 	"github.com/stretchr/testify/require"
 	"go.uber.org/zap"
@@ -790,10 +790,7 @@ func TestCacheWarmup(t *testing.T) {
 			})
 		})
 	})
-}
 
-// Is set as Flaky so that when running the tests it will be run separately and retried if it fails
-func TestFlakyCacheWarmupMetrics(t *testing.T) {
 	t.Run("should emit planning times metrics during warmup", func(t *testing.T) {
 		t.Parallel()
 
@@ -896,17 +893,36 @@ func TestFlakyCacheWarmupMetrics(t *testing.T) {
 			// One when warming up the operation and one when executing the operation
 			require.Len(t, m.Data.(metricdata.Histogram[float64]).DataPoints, 2)
 
+			// Find data-points by matching their cache hit attribute
+			warmupDataPoint := findDataPoint(t, m.Data.(metricdata.Histogram[float64]).DataPoints, false)
+			executionDataPoint := findDataPoint(t, m.Data.(metricdata.Histogram[float64]).DataPoints, true)
+
 			// Warming up the operation
-			require.Equal(t, m.Data.(metricdata.Histogram[float64]).DataPoints[0].Count, uint64(1))
+			require.Equal(t, uint64(1), warmupDataPoint.Count)
 
 			// Executing the operation. Is exported as an aggregated value
-			require.Equal(t, m.Data.(metricdata.Histogram[float64]).DataPoints[1].Count, uint64(2))
+			require.Equal(t, uint64(2), executionDataPoint.Count)
 
 			// Ensure we collected non-zero planning times
-			require.Greater(t, m.Data.(metricdata.Histogram[float64]).DataPoints[0].Sum, float64(0))
-			require.Greater(t, m.Data.(metricdata.Histogram[float64]).DataPoints[1].Sum, float64(0))
+			require.Greater(t, warmupDataPoint.Sum, float64(0))
+			require.Greater(t, executionDataPoint.Sum, float64(0))
 
 			metricdatatest.AssertEqual(t, operationPlanningTimeMetric, m, metricdatatest.IgnoreTimestamp(), metricdatatest.IgnoreValue())
 		})
 	})
 }
+
+// findDataPoint finds a data point in a slice of histogram data points by matching
+// the value of WgEnginePlanCacheHit attribute
+func findDataPoint(t *testing.T, dataPoints []metricdata.HistogramDataPoint[float64], cacheHit bool) metricdata.HistogramDataPoint[float64] {
+	t.Helper()
+	for _, dp := range dataPoints {
+		// Get the value of the WgEnginePlanCacheHit attribute
+		val, found := dp.Attributes.Value(otel.WgEnginePlanCacheHit)
+		if found && val.AsBool() == cacheHit {
+			return dp
+		}
+	}
+	t.Fatalf("Could not find data point with WgEnginePlanCacheHit=%v", cacheHit)
+	return metricdata.HistogramDataPoint[float64]{}
+}

router-tests/events/kafka_events_test.go

@@ -24,20 +24,6 @@ import (
 
 const KafkaWaitTimeout = time.Second * 30
 
-func TestLocalKafka(t *testing.T) {
-	t.Skip("skip only for local testing")
-
-	t.Run("subscribe async", func(t *testing.T) {
-		testenv.Run(t, &testenv.Config{
-			RouterConfigJSONTemplate: testenv.ConfigWithEdfsJSONTemplate,
-			EnableKafka:              true,
-		}, func(t *testing.T, xEnv *testenv.Environment) {
-			// ensureTopicExists(t, xEnv, "employeeUpdated", "employeeUpdatedTwo")
-			produceKafkaMessage(t, xEnv, "employeeUpdatedTwo", `{"__typename":"Employee","id": 2,"update":{"name":"foo"}}`)
-		})
-	})
-}
-
 func TestKafkaEvents(t *testing.T) {
 	t.Parallel()
 	// All tests are running in sequence because they are using the same kafka topic

router-tests/events/nats_events_test.go

@@ -386,7 +386,7 @@ func TestNatsEvents(t *testing.T) {
 			})
 		})
 
-		t.Run("subscribe with multipart responses http", func(t *testing.T) {
+		t.Run("subscribe with multipart responses http and consume healthcheck only", func(t *testing.T) {
 			t.Parallel()
 
 			testenv.Run(t, &testenv.Config{
@@ -402,12 +402,11 @@ func TestNatsEvents(t *testing.T) {
 
 				var counter atomic.Uint32
 
-				var client *http.Client
+				client := &http.Client{}
+
 				go func() {
 					defer counter.Add(1)
 
-					client = &http.Client{}
-
 					req := xEnv.MakeGraphQLMultipartRequest(http.MethodPost, bytes.NewReader(subscribePayload))
 					resp, err := client.Do(req)
 					require.NoError(t, err)
@@ -432,7 +431,6 @@ func TestNatsEvents(t *testing.T) {
 		})
 
 		t.Run("subscribe with closing channel", func(t *testing.T) {
-			t.Parallel()
 
 			testenv.Run(t, &testenv.Config{
 				RouterConfigJSONTemplate: testenv.ConfigWithEdfsNatsJSONTemplate,