merge: resolve conflict with main in gate_factory.go

Keep both the DefaultCacheTTL constant from this branch and the
GateFactory interface compliance check added on main.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: lioraron

pkg/async/api/api.go

@@ -41,6 +41,8 @@ type GateFactory interface {
 	CreateGate(gateType string, params map[string]string) (DispatchGate, error)
 }
 
+var _ DispatchGate = DispatchGateFunc(nil)
+
 // DispatchGateFunc is a function type that implements DispatchGate.
 // This allows any function with the signature func(context.Context) float64
 // to be used as a DispatchGate.

pkg/async/api/http_client.go

@@ -8,6 +8,8 @@ import (
 	"net/http"
 )
 
+var _ InferenceClient = (*HTTPInferenceClient)(nil)
+
 // HTTPInferenceClient is the default HTTP implementation of InferenceClient.
 type HTTPInferenceClient struct {
 	client *http.Client

pkg/async/api/inference_error.go

@@ -31,6 +31,8 @@ type InferenceError interface {
 	Category() ErrorCategory
 }
 
+var _ InferenceError = (*ClientError)(nil)
+
 // ClientError represents an inference client error with category and context.
 type ClientError struct {
 	ErrorCategory ErrorCategory

pkg/async/inference/flowcontrol/binary_metric_dispatch_gate.go

@@ -20,6 +20,7 @@ import (
 	"context"
 	"flag"
 
+	asyncapi "github.com/llm-d-incubation/llm-d-async/pkg/async/api"
 	"github.com/prometheus/client_golang/api"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
@@ -30,6 +31,8 @@ var prometheusURL = flag.String("gate.prometheus.url", "", "Prometheus URL for n
 var gmpProjectID = flag.String("gate.pmetric.gmp.project-id", "", "Project ID for Google Managed Prometheus")
 var prometheusQueryModelName = flag.String("gate.prometheus.model-name", "", "metrics name to use for avg_queue_size")
 
+var _ asyncapi.DispatchGate = (*BinaryMetricDispatchGate)(nil)
+
 // BinaryMetricDispatchGate implements DispatchGate using a MetricSource.
 // It returns 0.0 (no capacity) if the metric value is non-zero,
 // and 1.0 (full capacity) if the metric value is zero.

pkg/async/inference/flowcontrol/dispatch_gate.go

@@ -22,6 +22,8 @@ import (
 	"github.com/llm-d-incubation/llm-d-async/pkg/async/api"
 )
 
+var _ api.DispatchGate = DispatchGateFunc(nil)
+
 // DispatchGateFunc is a function type that implements DispatchGate.
 // This allows any function with the signature func(context.Context) float64
 // to be used as a DispatchGate.

pkg/async/inference/flowcontrol/gate_factory.go

@@ -30,6 +30,8 @@ import (
 // DefaultCacheTTL is the default TTL for cached Prometheus metric sources.
 const DefaultCacheTTL = 5 * time.Second
 
+var _ asyncapi.GateFactory = (*GateFactory)(nil)
+
 // GateFactory creates DispatchGate instances based on configuration.
 type GateFactory struct {
 	prometheusURL string

pkg/async/inference/flowcontrol/metric_source.go

@@ -46,6 +46,8 @@ type MetricSource interface {
 	Query(ctx context.Context) ([]Sample, error)
 }
 
+var _ MetricSource = (*PromQLMetricSource)(nil)
+
 // PromQLMetricSource implements MetricSource by executing a PromQL expression
 // against a Prometheus-compatible API.
 type PromQLMetricSource struct {

pkg/async/inference/flowcontrol/saturation_metric_dispatch_gate.go

@@ -21,6 +21,7 @@ import (
 	"flag"
 	"math"
 
+	asyncapi "github.com/llm-d-incubation/llm-d-async/pkg/async/api"
 	"github.com/prometheus/client_golang/api"
 	"sigs.k8s.io/controller-runtime/pkg/log"
 	logutil "sigs.k8s.io/gateway-api-inference-extension/pkg/epp/util/logging"
@@ -31,6 +32,8 @@ var saturationThreshold = flag.Float64("gate.saturation.threshold", 0.8, "satura
 var saturationFallback = flag.Float64("gate.saturation.fallback", 0.0, "fallback saturation value on error/missing metrics; default 0.0")
 var saturationQueryExpr = flag.String("gate.saturation.query-expr", "", "custom PromQL expression for saturation metric; overrides inference-pool label selector")
 
+var _ asyncapi.DispatchGate = (*SaturationMetricDispatchGate)(nil)
+
 // SaturationMetricDispatchGate implements DispatchGate based on pool saturation.
 // It queries a MetricSource for saturation samples and returns 0.0 if saturation
 // is at or above the configured threshold, otherwise returns max(0, 1 - saturation),

pkg/async/random_robin_policy.go

@@ -10,11 +10,13 @@ func NewRandomRobinPolicy() api.RequestMergePolicy {
 	return &RandomRobinPolicy{}
 }
 
+var _ api.RequestMergePolicy = (*RandomRobinPolicy)(nil)
+
 type RandomRobinPolicy struct {
 }
 
 func (r *RandomRobinPolicy) MergeRequestChannels(channels []api.RequestChannel) api.EmbelishedRequestChannel {
-	mergedChannel := make(chan api.EmbelishedRequestMessage)
+	mergedChannel := make(chan api.EmbelishedRequestMessage, len(channels))
 
 	cases := make([]reflect.SelectCase, len(channels)) //nolint:staticcheck
 	for i, ch := range channels {

pkg/async/random_robin_policy_test.go

@@ -2,6 +2,7 @@ package async
 
 import (
 	"testing"
+	"time"
 
 	"github.com/llm-d-incubation/llm-d-async/pkg/async/api"
 )
@@ -41,3 +42,33 @@ func TestProcessAllChannels(t *testing.T) {
 		}
 	}
 }
+
+func TestMergedChannelIsBuffered(t *testing.T) {
+	numChannels := 3
+	channels := make([]api.RequestChannel, numChannels)
+	for i := range numChannels {
+		channels[i] = api.RequestChannel{Channel: make(chan api.RequestMessage, 1)}
+	}
+	policy := NewRandomRobinPolicy()
+	merged := policy.MergeRequestChannels(channels)
+
+	// Send one message per input channel.
+	for i, ch := range channels {
+		ch.Channel <- api.RequestMessage{Id: string(rune('A' + i))}
+	}
+
+	// The merge goroutine should be able to forward all messages into the
+	// buffered merged channel without a consumer draining it. With an
+	// unbuffered channel this would deadlock because the goroutine blocks
+	// on the first send.
+	deadline := time.After(2 * time.Second)
+	received := 0
+	for received < numChannels {
+		select {
+		case <-merged.Channel:
+			received++
+		case <-deadline:
+			t.Fatalf("timed out: only received %d/%d messages — merged channel may be unbuffered", received, numChannels)
+		}
+	}
+}