feat: add per-plugin semaphore_size and inject_timeout to PluginConfig with context-bounded Inject calls and tracer-default fallbacks

core/schemas/plugin.go

@@ -5,6 +5,7 @@ import (
 	"context"
 	"strings"
 	"sync"
+	"time"
 )
 
 // PluginStatus constants
@@ -368,6 +369,17 @@ type PluginConfig struct {
 	Config    any              `json:"config,omitempty"`
 	Placement *PluginPlacement `json:"placement,omitempty"` // "pre_builtin" or "post_builtin". Default: "post_builtin"
 	Order     *int             `json:"order,omitempty"`     // Position within placement group. Lower = earlier. Default: 0
+
+	// SemaphoreSize caps concurrent in-flight Inject calls the tracer will send this
+	// plugin, if it implements ObservabilityPlugin. Generic (like Enabled) rather than
+	// part of each plugin's own Config, since the tracer holds one semaphore per plugin
+	// regardless of that plugin's internal shape. Zero/unset falls back to the tracer's
+	// default (10000). Ignored by plugins that don't implement ObservabilityPlugin.
+	SemaphoreSize *int `json:"semaphore_size,omitempty"`
+	// InjectTimeout bounds a single Inject call, as a Go duration string (e.g. "5s").
+	// Generic for the same reason as SemaphoreSize. Zero/unset falls back to the
+	// tracer's default (5s). Ignored by plugins that don't implement ObservabilityPlugin.
+	InjectTimeout *string `json:"inject_timeout,omitempty"`
 }
 
 // ConfigMarshallerPlugin is optionally implemented by plugins that need custom
@@ -414,11 +426,29 @@ type ObservabilityPlugin interface {
 	// - Send the trace to the backend (can be async, but see retention note below)
 	// - Handle errors gracefully (log and continue)
 	//
-	// The context passed is a fresh background context, not the request context.
+	// The context passed is derived from a fresh background context (not the request
+	// context), bounded by the plugin's inject timeout (see ObservabilityLimits).
+	// Implementations that perform network I/O should propagate ctx into their client
+	// calls so a hung backend is actually unblocked when the timeout fires, rather than
+	// leaking the goroutine and its concurrency-budget slot indefinitely.
 	//
 	// Retention: implementations MUST NOT retain the *Trace pointer after Inject
 	// returns. The caller releases the underlying trace back to a sync.Pool
 	// immediately after Inject completes. If a plugin needs to forward the trace
 	// asynchronously, it must copy the data it needs before returning.
 	Inject(ctx context.Context, trace *Trace) error
 }
+
+// ObservabilityLimits configures how the tracer bounds a single observability plugin's
+// Inject calls: how many may run concurrently, and how long any one call is allowed to take.
+// Resolved from the plugin's generic PluginConfig.SemaphoreSize/InjectTimeout — plugins
+// themselves have no say in this, the same way they don't decide their own Enabled state.
+type ObservabilityLimits struct {
+	// SemaphoreSize caps concurrent in-flight Inject calls for this plugin. A trace is
+	// dropped for this plugin (not process-wide) when the cap is already saturated.
+	// Zero/unset falls back to the tracer's default (10000).
+	SemaphoreSize int
+	// InjectTimeout bounds a single Inject call. Zero/unset falls back to the tracer's
+	// default (5s).
+	InjectTimeout time.Duration
+}

framework/tracing/obsisolation_test.go

@@ -76,7 +76,7 @@ func TestCompleteAndFlushTrace_SlowPluginDoesNotDelayOthers(t *testing.T) {
 	fast := &blockingObsPlugin{name: "fast-connector"}
 
 	// Slow one registered first: the ordering that used to cause the stall.
-	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{slow, fast})
+	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{slow, fast}, nil)
 
 	traceID := tracer.CreateTrace("")
 	start := time.Now()
@@ -106,10 +106,10 @@ func TestCompleteAndFlushTrace_BoundsInjectsPerPlugin(t *testing.T) {
 	tracer := NewTracer(store, nil, nil)
 
 	stuck := &blockingObsPlugin{name: "stuck-connector", release: make(chan struct{})}
-	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{stuck})
+	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{stuck}, nil)
 
 	const overshoot = 250
-	total := maxConcurrentInjectsPerPlugin + overshoot
+	total := defaultSemaphoreSize + overshoot
 	for range total {
 		tracer.CompleteAndFlushTrace(tracer.CreateTrace(""))
 	}
@@ -120,8 +120,8 @@ func TestCompleteAndFlushTrace_BoundsInjectsPerPlugin(t *testing.T) {
 		time.Sleep(5 * time.Millisecond)
 	}
 
-	if got := stuck.maxInFlight.Load(); got > maxConcurrentInjectsPerPlugin {
-		t.Fatalf("concurrent injects exceeded the cap: got %d, cap %d", got, maxConcurrentInjectsPerPlugin)
+	if got := stuck.maxInFlight.Load(); got > defaultSemaphoreSize {
+		t.Fatalf("concurrent injects exceeded the cap: got %d, cap %d", got, defaultSemaphoreSize)
 	}
 	if dropped := tracer.ObservabilityDropCounts()["stuck-connector"]; dropped == 0 {
 		t.Fatal("expected traces to be skipped once the plugin saturated, got 0")
@@ -140,7 +140,7 @@ func TestWaitForFlushes_TimesOutOnHungPlugin(t *testing.T) {
 	tracer := NewTracer(store, nil, nil)
 
 	hung := &blockingObsPlugin{name: "hung-connector", release: make(chan struct{})}
-	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{hung})
+	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{hung}, nil)
 	tracer.CompleteAndFlushTrace(tracer.CreateTrace(""))
 
 	deadline := time.Now().Add(5 * time.Second)
@@ -173,7 +173,7 @@ func TestSetObservabilityPlugins_DedupesByName(t *testing.T) {
 	defer tracer.Stop()
 
 	dup := &blockingObsPlugin{name: "dup-connector"}
-	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{dup, dup, dup})
+	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{dup, dup, dup}, nil)
 
 	var wg sync.WaitGroup
 	wg.Add(1)
@@ -193,3 +193,131 @@ func TestSetObservabilityPlugins_DedupesByName(t *testing.T) {
 		t.Fatalf("expected a duplicate-named plugin to be injected once, got %d", got)
 	}
 }
+
+// ctxAwareObsPlugin's Inject respects ctx cancellation: it blocks until either released or
+// the context is done, whichever comes first. Stands in for a well-behaved connector whose
+// HTTP/gRPC client propagates ctx.
+type ctxAwareObsPlugin struct {
+	name    string
+	release chan struct{}
+	started atomic.Int64
+}
+
+func (p *ctxAwareObsPlugin) GetName() string { return p.name }
+func (p *ctxAwareObsPlugin) Cleanup() error  { return nil }
+func (p *ctxAwareObsPlugin) Inject(ctx context.Context, _ *schemas.Trace) error {
+	p.started.Add(1)
+	select {
+	case <-p.release:
+		return nil
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+// TestSetObservabilityPlugins_HonoursDeclaredLimits verifies a plugin whose generic
+// PluginConfig declares semaphore_size/inject_timeout gets a semaphore sized from that
+// config instead of the tracer's default.
+func TestSetObservabilityPlugins_HonoursDeclaredLimits(t *testing.T) {
+	store := NewTraceStore(5*time.Minute, nil)
+	defer store.Stop()
+
+	tracer := NewTracer(store, nil, nil)
+	defer tracer.Stop()
+
+	const customSemSize = 4
+	plugin := &blockingObsPlugin{name: "limited-connector", release: make(chan struct{})}
+	limits := map[string]schemas.ObservabilityLimits{
+		"limited-connector": {SemaphoreSize: customSemSize, InjectTimeout: time.Minute},
+	}
+	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{plugin}, limits)
+
+	loaded := tracer.obsPlugins.Load()
+	if loaded == nil || len(*loaded) != 1 {
+		t.Fatalf("expected exactly one slot, got %v", loaded)
+	}
+	slot := (*loaded)[0]
+	if cap(slot.sem) != customSemSize {
+		t.Fatalf("expected semaphore sized %d from declared limits, got %d", customSemSize, cap(slot.sem))
+	}
+	if slot.injectTimeout != time.Minute {
+		t.Fatalf("expected inject timeout from declared limits, got %v", slot.injectTimeout)
+	}
+
+	close(plugin.release)
+}
+
+// TestSetObservabilityPlugins_DefaultsWhenLimitsNotDeclared verifies a plugin with no
+// entry in the limits map falls back to the tracer's defaults.
+func TestSetObservabilityPlugins_DefaultsWhenLimitsNotDeclared(t *testing.T) {
+	store := NewTraceStore(5*time.Minute, nil)
+	defer store.Stop()
+
+	tracer := NewTracer(store, nil, nil)
+	defer tracer.Stop()
+
+	plain := &blockingObsPlugin{name: "plain-connector"}
+	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{plain}, nil)
+
+	loaded := tracer.obsPlugins.Load()
+	if loaded == nil || len(*loaded) != 1 {
+		t.Fatalf("expected exactly one slot, got %v", loaded)
+	}
+	slot := (*loaded)[0]
+	if cap(slot.sem) != defaultSemaphoreSize {
+		t.Fatalf("expected default semaphore size %d, got %d", defaultSemaphoreSize, cap(slot.sem))
+	}
+	if slot.injectTimeout != defaultInjectTimeout {
+		t.Fatalf("expected default inject timeout %v, got %v", defaultInjectTimeout, slot.injectTimeout)
+	}
+}
+
+// TestCompleteAndFlushTrace_InjectTimeoutReleasesSlot verifies that a plugin honouring ctx
+// cancellation has its Inject call unblocked once the configured inject timeout elapses,
+// freeing the semaphore slot instead of holding it indefinitely.
+func TestCompleteAndFlushTrace_InjectTimeoutReleasesSlot(t *testing.T) {
+	store := NewTraceStore(5*time.Minute, nil)
+	defer store.Stop()
+
+	tracer := NewTracer(store, nil, nil)
+	defer tracer.Stop()
+
+	// release is never closed: the plugin only returns because its 50ms inject timeout
+	// (declared via the limits map passed to SetObservabilityPlugins) cancels ctx.
+	plugin := &ctxAwareObsPlugin{name: "ctx-aware-connector", release: make(chan struct{})}
+	// SemaphoreSize: 1 makes the test meaningful — with the tracer's default of 10,000,
+	// both flushes below would acquire a slot immediately regardless of whether
+	// cancellation actually released one.
+	limits := map[string]schemas.ObservabilityLimits{
+		"ctx-aware-connector": {SemaphoreSize: 1, InjectTimeout: 50 * time.Millisecond},
+	}
+	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{plugin}, limits)
+
+	// SemaphoreSize is 1, so the second flush can only start once the first Inject's
+	// ctx cancellation frees the slot. Submitting and waiting one at a time (rather
+	// than firing both up front) is what actually exercises that release path — with
+	// both queued together, the second could otherwise sit dropped by the non-blocking
+	// acquire instead of proving the slot came free.
+	start := time.Now()
+	tracer.CompleteAndFlushTrace(tracer.CreateTrace(""))
+	if completed := tracer.waitForFlushes(2 * time.Second); !completed {
+		t.Fatal("expected the first flush to complete once its inject timeout elapsed")
+	}
+	if got := plugin.started.Load(); got != 1 {
+		t.Fatalf("expected first flush to start Inject once, got %d", got)
+	}
+
+	tracer.CompleteAndFlushTrace(tracer.CreateTrace(""))
+	if completed := tracer.waitForFlushes(2 * time.Second); !completed {
+		t.Fatal("expected the second flush to complete once its inject timeout elapsed")
+	}
+	if got := plugin.started.Load(); got != 2 {
+		t.Fatalf("expected the second flush's Inject to start once the first slot was released, got %d starts", got)
+	}
+	if dropped := tracer.ObservabilityDropCounts()["ctx-aware-connector"]; dropped != 0 {
+		t.Fatalf("expected no traces dropped, got %d", dropped)
+	}
+	if elapsed := time.Since(start); elapsed > time.Second {
+		t.Fatalf("flushes took %v, expected them to be bounded by the ~50ms inject timeout each", elapsed)
+	}
+}

framework/tracing/tracer.go

@@ -3,6 +3,7 @@ package tracing
 
 import (
 	"context"
+	"errors"
 	"strings"
 	"sync"
 	"sync/atomic"
@@ -14,28 +15,51 @@ import (
 )
 
 const (
-	// maxConcurrentInjectsPerPlugin bounds how many traces may sit inside a single
-	// observability connector's Inject at once. A connector pointed at a wrong or
-	// black-holed endpoint blocks on network I/O while pinning a full trace snapshot;
-	// uncapped that is one goroutine and one snapshot per request, whose heap and
-	// scheduler pressure starves the rest of the process — notably the logging
-	// plugin's single DB batch writer, whose queue then drops rows silently.
+	// defaultSemaphoreSize bounds how many traces may sit inside a single observability
+	// connector's Inject at once when its PluginConfig doesn't set semaphore_size. A
+	// connector pointed at a wrong or black-holed endpoint blocks on network I/O while
+	// pinning a full trace snapshot; uncapped that is one goroutine and one snapshot per
+	// request, whose heap and scheduler pressure starves the rest of the process —
+	// notably the logging plugin's single DB batch writer, whose queue then drops rows
+	// silently.
 	// Each connector gets its own budget so a stalled one cannot crowd out a healthy one.
-	maxConcurrentInjectsPerPlugin = 1024
+	defaultSemaphoreSize = 10000
+
+	// defaultInjectTimeout bounds a single Inject call when its PluginConfig doesn't set
+	// inject_timeout. Paired with the per-plugin semaphore, this is what actually
+	// releases a slot when a connector hangs — the semaphore alone only limits how many
+	// hung calls can pile up, not how long each does.
+	defaultInjectTimeout = 5 * time.Second
 
 	// flushStopTimeout bounds how long Stop waits for in-flight trace exports, so a
 	// hung collector cannot block shutdown or a config reload.
 	flushStopTimeout = 10 * time.Second
 )
 
-// obsPluginSlot pairs an observability plugin with its own concurrency budget and
-// drop counter. Isolating the budget per connector is what keeps a misconfigured
-// exporter from consuming the capacity that healthy connectors need.
+// obsPluginSlot pairs an observability plugin with its own concurrency budget, inject
+// timeout, and drop counter. Isolating the budget per connector is what keeps a
+// misconfigured exporter from consuming the capacity that healthy connectors need.
 type obsPluginSlot struct {
-	plugin  schemas.ObservabilityPlugin
-	name    string
-	sem     chan struct{}
-	dropped atomic.Int64
+	plugin        schemas.ObservabilityPlugin
+	name          string
+	sem           chan struct{}
+	injectTimeout time.Duration
+	dropped       atomic.Int64
+}
+
+// resolveObservabilityLimits applies the tracer defaults to whatever limits a plugin's
+// generic PluginConfig declared (see schemas.PluginConfig.SemaphoreSize/InjectTimeout).
+// A zero field means "unset", not "zero" — it falls back to the default rather than
+// producing a zero-size semaphore or an immediately-expiring timeout.
+func resolveObservabilityLimits(limits schemas.ObservabilityLimits) (semSize int, injectTimeout time.Duration) {
+	semSize, injectTimeout = defaultSemaphoreSize, defaultInjectTimeout
+	if limits.SemaphoreSize > 0 {
+		semSize = limits.SemaphoreSize
+	}
+	if limits.InjectTimeout > 0 {
+		injectTimeout = limits.InjectTimeout
+	}
+	return semSize, injectTimeout
 }
 
 // Tracer implements schemas.Tracer using TraceStore.
@@ -65,10 +89,13 @@ func NewTracer(store *TraceStore, pricingManager *modelcatalog.ModelCatalog, log
 	}
 }
 
-// SetObservabilityPlugins updates the plugins that receive completed traces.
+// SetObservabilityPlugins updates the plugins that receive completed traces. limits is
+// keyed by plugin name (GetName()), sourced from each plugin's generic PluginConfig
+// (schemas.PluginConfig.SemaphoreSize/InjectTimeout) — a plugin absent from the map, or
+// with unset fields, gets the tracer's defaults.
 // It also precomputes the deduplicated, normalized union of request-header patterns
 // requested by those plugins so the per-request capture path is a single atomic load.
-func (t *Tracer) SetObservabilityPlugins(obsPlugins []schemas.ObservabilityPlugin) {
+func (t *Tracer) SetObservabilityPlugins(obsPlugins []schemas.ObservabilityPlugin, limits map[string]schemas.ObservabilityLimits) {
 	if t == nil {
 		return
 	}
@@ -86,10 +113,12 @@ func (t *Tracer) SetObservabilityPlugins(obsPlugins []schemas.ObservabilityPlugi
 			continue
 		}
 		seenPlugins[name] = struct{}{}
+		semSize, injectTimeout := resolveObservabilityLimits(limits[name])
 		slots = append(slots, &obsPluginSlot{
-			plugin: plugin,
-			name:   name,
-			sem:    make(chan struct{}, maxConcurrentInjectsPerPlugin),
+			plugin:        plugin,
+			name:          name,
+			sem:           make(chan struct{}, semSize),
+			injectTimeout: injectTimeout,
 		})
 	}
 	t.obsPlugins.Store(&slots)
@@ -858,7 +887,7 @@ func (t *Tracer) CompleteAndFlushTrace(traceID string) {
 				// the logging itself a second source of load.
 				if n := slot.dropped.Add(1); (n == 1 || n%1000 == 0) && t.logger != nil {
 					t.logger.Warn("observability plugin %s saturated at %d concurrent injects, skipped trace %s (%d skipped so far)",
-						slot.name, maxConcurrentInjectsPerPlugin, exportTrace.TraceID, n)
+						slot.name, cap(slot.sem), exportTrace.TraceID, n)
 				}
 				continue
 			}
@@ -873,8 +902,14 @@ func (t *Tracer) CompleteAndFlushTrace(traceID string) {
 						t.logger.Error("observability plugin %s panicked during trace injection: %v", slot.name, r)
 					}
 				}()
-				if err := slot.plugin.Inject(context.Background(), exportTrace); err != nil && t.logger != nil {
-					t.logger.Warn("observability plugin %s failed to inject trace: %v", slot.name, err)
+				injectCtx, cancel := context.WithTimeout(context.Background(), slot.injectTimeout)
+				defer cancel()
+				if err := slot.plugin.Inject(injectCtx, exportTrace); err != nil && t.logger != nil {
+					if errors.Is(err, context.DeadlineExceeded) {
+						t.logger.Warn("observability plugin %s timed out injecting trace %s after %s", slot.name, exportTrace.TraceID, slot.injectTimeout)
+					} else {
+						t.logger.Warn("observability plugin %s failed to inject trace: %v", slot.name, err)
+					}
 				}
 			}(slot)
 		}

framework/tracing/tracer_test.go

@@ -60,7 +60,7 @@ func TestTracer_CompleteAndFlushTraceInjectsObservabilityPlugins(t *testing.T) {
 		injected: make(chan *schemas.Trace, 1),
 	}
 
-	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{plugin})
+	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{plugin}, nil)
 	tracer.CompleteAndFlushTrace(traceID)
 
 	select {
@@ -88,7 +88,7 @@ func TestTracer_CompleteAndFlushTraceRedactsContentBeforeInject(t *testing.T) {
 	plugin := &testRealtimeObservabilityPlugin{
 		injectedPayload: make(chan string, 1),
 	}
-	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{plugin})
+	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{plugin}, nil)
 
 	// Store replacements before output attributes are populated. This mirrors
 	// streaming, where the final accumulated output lands near trace completion.
@@ -142,7 +142,7 @@ func TestTracer_SetTraceRedactionReplacementsSurvivesLaterObservabilityPlugins(t
 	plugin := &testRealtimeObservabilityPlugin{
 		injectedPayload: make(chan string, 1),
 	}
-	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{plugin})
+	tracer.SetObservabilityPlugins([]schemas.ObservabilityPlugin{plugin}, nil)
 
 	ctx := context.WithValue(context.Background(), schemas.BifrostContextKeyTraceID, traceID)
 	_, rootHandle := tracer.StartSpan(ctx, "http-request", schemas.SpanKindHTTPRequest)