feat: add request queue limits

README.md

@@ -30,6 +30,11 @@ This repository is in early foundation work. The current service supports:
 Routing policy, auth, telemetry, LM Studio lifecycle integration, and Omarchy
 integration are planned next.
 
+Model aliases can also set basic concurrency guardrails with
+`max_concurrent_requests`, `max_queue_size`, and `queue_timeout`. This lets heavy
+local models wait or reject predictably instead of allowing multiple agents to
+dogpile the same backend.
+
 ## Quick Start
 
 Build and test locally:
@@ -98,6 +103,9 @@ models:
     context_window: 65536
     max_output_tokens: 4096
     tool_calls: true
+    max_concurrent_requests: 2
+    max_queue_size: 4
+    queue_timeout: 30s
 
 backends:
   - id: lmstudio

configs/router.docker.yaml

@@ -9,13 +9,19 @@ models:
     context_window: 65536
     max_output_tokens: 4096
     tool_calls: true
+    max_concurrent_requests: 2
+    max_queue_size: 4
+    queue_timeout: 30s
   - id: local-coder-large
     name: Local Coder Large
     backend: mock-openai
     target_model: qwen/qwen3.6-35b-a3b
     context_window: 131072
     max_output_tokens: 4096
     tool_calls: true
+    max_concurrent_requests: 1
+    max_queue_size: 2
+    queue_timeout: 2m
 
 backends:
   - id: mock-openai

configs/router.example.yaml

@@ -9,13 +9,19 @@ models:
     context_window: 65536
     max_output_tokens: 4096
     tool_calls: true
+    max_concurrent_requests: 2
+    max_queue_size: 4
+    queue_timeout: 30s
   - id: local-coder-large
     name: Local Coder Large
     backend: lmstudio
     target_model: qwen/qwen3.6-35b-a3b
     context_window: 131072
     max_output_tokens: 4096
     tool_calls: true
+    max_concurrent_requests: 1
+    max_queue_size: 2
+    queue_timeout: 2m
 
 backends:
   - id: lmstudio

docs/architecture.md

@@ -37,3 +37,28 @@ The first router is deliberately simple:
 
 Future routing can add deterministic policy, queueing, health-aware selection,
 RouteLLM-style strong/weak model routing, and second-pass review workflows.
+
+## Request Limits
+
+Model aliases can define optional concurrency and queue limits:
+
+```yaml
+models:
+  - id: local-coder-large
+    backend: lmstudio
+    target_model: qwen/qwen3.6-35b-a3b
+    max_concurrent_requests: 1
+    max_queue_size: 2
+    queue_timeout: 2m
+```
+
+When `max_concurrent_requests` is unset or `0`, the alias is unlimited. When it
+is set, DevRail Router holds one slot for each proxied request until the
+upstream response is fully complete. That matters for streaming chat responses:
+the slot is not released while tokens are still flowing.
+
+If all slots are busy, requests can wait in the bounded queue. If the queue is
+full, DevRail Router returns an OpenAI-shaped `429` error. If the request waits
+longer than `queue_timeout`, it returns `503`. Successful queued requests get an
+`X-Devrail-Queue-Wait-Ms` response header, and queue decisions are logged with
+active count, queued count, and wait time.

internal/config/config.go

@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"net/url"
 	"os"
+	"time"
 
 	"gopkg.in/yaml.v3"
 )
@@ -22,13 +23,16 @@ type ServerConfig struct {
 }
 
 type ModelConfig struct {
-	ID              string `yaml:"id"`
-	Name            string `yaml:"name"`
-	Backend         string `yaml:"backend"`
-	TargetModel     string `yaml:"target_model"`
-	ContextWindow   int    `yaml:"context_window"`
-	MaxOutputTokens int    `yaml:"max_output_tokens"`
-	ToolCalls       bool   `yaml:"tool_calls"`
+	ID                    string `yaml:"id"`
+	Name                  string `yaml:"name"`
+	Backend               string `yaml:"backend"`
+	TargetModel           string `yaml:"target_model"`
+	ContextWindow         int    `yaml:"context_window"`
+	MaxOutputTokens       int    `yaml:"max_output_tokens"`
+	ToolCalls             bool   `yaml:"tool_calls"`
+	MaxConcurrentRequests int    `yaml:"max_concurrent_requests"`
+	MaxQueueSize          int    `yaml:"max_queue_size"`
+	QueueTimeout          string `yaml:"queue_timeout"`
 }
 
 type BackendConfig struct {
@@ -102,6 +106,15 @@ func (cfg Config) Validate() error {
 		if model.TargetModel == "" {
 			return fmt.Errorf("model %q target_model is required", model.ID)
 		}
+		if model.MaxConcurrentRequests < 0 {
+			return fmt.Errorf("model %q max_concurrent_requests must be non-negative", model.ID)
+		}
+		if model.MaxQueueSize < 0 {
+			return fmt.Errorf("model %q max_queue_size must be non-negative", model.ID)
+		}
+		if _, err := model.QueueTimeoutDuration(); err != nil {
+			return fmt.Errorf("model %q queue_timeout is invalid: %w", model.ID, err)
+		}
 		if _, ok := models[model.ID]; ok {
 			return fmt.Errorf("model %q is duplicated", model.ID)
 		}
@@ -130,3 +143,19 @@ func (cfg Config) Backend(id string) (BackendConfig, bool) {
 
 	return BackendConfig{}, false
 }
+
+func (model ModelConfig) QueueTimeoutDuration() (time.Duration, error) {
+	if model.QueueTimeout == "" {
+		return 0, nil
+	}
+
+	duration, err := time.ParseDuration(model.QueueTimeout)
+	if err != nil {
+		return 0, err
+	}
+	if duration < 0 {
+		return 0, errors.New("duration must be non-negative")
+	}
+
+	return duration, nil
+}

internal/config/config_test.go

@@ -3,7 +3,9 @@ package config
 import (
 	"os"
 	"path/filepath"
+	"strings"
 	"testing"
+	"time"
 )
 
 func TestLoadValidConfig(t *testing.T) {
@@ -37,6 +39,62 @@ backends:
 	}
 }
 
+func TestValidateQueueSettings(t *testing.T) {
+	t.Parallel()
+
+	cfg := Config{
+		Models: []ModelConfig{{
+			ID:                    "local-coder",
+			Backend:               "lmstudio",
+			TargetModel:           "qwen/qwen3.6-35b-a3b",
+			MaxConcurrentRequests: 1,
+			MaxQueueSize:          2,
+			QueueTimeout:          "250ms",
+		}},
+		Backends: []BackendConfig{{
+			ID:      "lmstudio",
+			BaseURL: "http://127.0.0.1:1234/v1",
+		}},
+	}
+
+	if err := cfg.Validate(); err != nil {
+		t.Fatalf("validate config: %v", err)
+	}
+
+	duration, err := cfg.Models[0].QueueTimeoutDuration()
+	if err != nil {
+		t.Fatalf("parse queue timeout: %v", err)
+	}
+	if duration != 250*time.Millisecond {
+		t.Fatalf("unexpected queue timeout: %s", duration)
+	}
+}
+
+func TestValidateInvalidQueueTimeout(t *testing.T) {
+	t.Parallel()
+
+	cfg := Config{
+		Models: []ModelConfig{{
+			ID:           "local-coder",
+			Backend:      "lmstudio",
+			TargetModel:  "qwen/qwen3.6-35b-a3b",
+			QueueTimeout: "eventually",
+		}},
+		Backends: []BackendConfig{{
+			ID:      "lmstudio",
+			BaseURL: "http://127.0.0.1:1234/v1",
+		}},
+	}
+
+	err := cfg.Validate()
+	if err == nil {
+		t.Fatal("expected validation error")
+	}
+	if !strings.Contains(err.Error(), "queue_timeout") {
+		t.Fatalf("expected queue_timeout error, got: %v", err)
+	}
+}
+
 func TestValidateUnknownBackend(t *testing.T) {
 	t.Parallel()
 

internal/server/limiter.go

@@ -0,0 +1,125 @@
+package server
+
+import (
+	"context"
+	"errors"
+	"sync"
+	"time"
+
+	"github.com/devrail-dev/devrail-router/internal/config"
+)
+
+var (
+	errQueueFull    = errors.New("model queue is full")
+	errQueueTimeout = errors.New("timed out waiting for model queue")
+)
+
+type modelLimiter struct {
+	modelID      string
+	slots        chan struct{}
+	maxQueueSize int
+	queueTimeout time.Duration
+
+	mu     sync.Mutex
+	active int
+	queued int
+}
+
+type limiterSnapshot struct {
+	active int
+	queued int
+}
+
+func newModelLimiter(model config.ModelConfig) (*modelLimiter, error) {
+	queueTimeout, err := model.QueueTimeoutDuration()
+	if err != nil {
+		return nil, err
+	}
+	if model.MaxConcurrentRequests <= 0 {
+		return nil, nil
+	}
+
+	return &modelLimiter{
+		modelID:      model.ID,
+		slots:        make(chan struct{}, model.MaxConcurrentRequests),
+		maxQueueSize: model.MaxQueueSize,
+		queueTimeout: queueTimeout,
+	}, nil
+}
+
+func (limiter *modelLimiter) acquire(ctx context.Context) (time.Duration, limiterSnapshot, func(), error) {
+	started := time.Now()
+
+	select {
+	case limiter.slots <- struct{}{}:
+		snapshot := limiter.incrementActive()
+		return 0, snapshot, limiter.release, nil
+	default:
+	}
+
+	if !limiter.joinQueue() {
+		return 0, limiter.snapshot(), nil, errQueueFull
+	}
+	defer limiter.leaveQueue()
+
+	waitCtx := ctx
+	cancel := func() {}
+	if limiter.queueTimeout > 0 {
+		waitCtx, cancel = context.WithTimeout(ctx, limiter.queueTimeout)
+	}
+	defer cancel()
+
+	select {
+	case limiter.slots <- struct{}{}:
+		waited := time.Since(started)
+		snapshot := limiter.incrementActive()
+		return waited, snapshot, limiter.release, nil
+	case <-waitCtx.Done():
+		if errors.Is(waitCtx.Err(), context.DeadlineExceeded) {
+			return time.Since(started), limiter.snapshot(), nil, errQueueTimeout
+		}
+		return time.Since(started), limiter.snapshot(), nil, waitCtx.Err()
+	}
+}
+
+func (limiter *modelLimiter) joinQueue() bool {
+	limiter.mu.Lock()
+	defer limiter.mu.Unlock()
+
+	if limiter.queued >= limiter.maxQueueSize {
+		return false
+	}
+	limiter.queued++
+	return true
+}
+
+func (limiter *modelLimiter) leaveQueue() {
+	limiter.mu.Lock()
+	defer limiter.mu.Unlock()
+
+	limiter.queued--
+}
+
+func (limiter *modelLimiter) incrementActive() limiterSnapshot {
+	limiter.mu.Lock()
+	defer limiter.mu.Unlock()
+
+	limiter.active++
+	return limiterSnapshot{active: limiter.active, queued: limiter.queued}
+}
+
+func (limiter *modelLimiter) release() {
+	<-limiter.slots
+
+	limiter.mu.Lock()
+	defer limiter.mu.Unlock()
+
+	limiter.active--
+}
+
+func (limiter *modelLimiter) snapshot() limiterSnapshot {
+	limiter.mu.Lock()
+	defer limiter.mu.Unlock()
+
+	return limiterSnapshot{active: limiter.active, queued: limiter.queued}
+}