feat(bulkhead): model queued connection acquisition

src/components/Inspector.tsx

@@ -96,6 +96,9 @@ type Field =
   | 'intervalMs'
   | 'batchSize'
   | 'bulkheadMax'
+  | 'bulkheadMode'
+  | 'acquireQueueMax'
+  | 'acquireTimeoutMs'
   | 'flushDelayMs'
   | 'edgeShare'
   | 'lowPriorityShare'
@@ -288,7 +291,7 @@ const FIELDS_BY_KIND: Record<NodeKind, Field[]> = {
   // One knob, and it IS the component: how many calls may be outstanding
   // to the dependency behind it. errorRate and retries are deliberately
   // not offered here; both would make the pool count drift.
-  bulkhead: ['bulkheadMax'],
+  bulkhead: ['bulkheadMax', 'bulkheadMode', 'acquireQueueMax', 'acquireTimeoutMs'],
   // Delivery concurrency, dispatch cost, buffer depth, and the redrive
   // policy: attempts and the per-attempt deadline.
   retryqueue: [
@@ -496,7 +499,15 @@ interface NumberSpec {
   step: number;
 }
 
-type FieldSpec = SliderSpec | NumberSpec;
+interface ChoiceSpec {
+  control: 'choice';
+  label: string;
+  unit: string;
+  hint?: FieldHint;
+  options: readonly { value: 'reject' | 'wait'; label: string }[];
+}
+
+type FieldSpec = SliderSpec | NumberSpec | ChoiceSpec;
 
 const FIELD_SPECS: Record<Field, FieldSpec> = {
   rps: {
@@ -973,6 +984,33 @@ const FIELD_SPECS: Record<Field, FieldSpec> = {
     max: 512,
     step: 1,
   },
+  bulkheadMode: {
+    control: 'choice',
+    label: 'When the pool is full',
+    unit: 'admission behavior',
+    hint: 'Reject fails immediately. Wait holds a request until a connection opens, or its acquire timeout expires.',
+    options: [
+      { value: 'reject', label: 'Reject' },
+      { value: 'wait', label: 'Wait' },
+    ],
+  },
+  acquireQueueMax: {
+    control: 'number',
+    label: 'Waiting acquires, at most',
+    unit: 'requests',
+    min: 0,
+    max: 10000,
+    step: 1,
+  },
+  acquireTimeoutMs: {
+    control: 'slider',
+    label: 'Acquire timeout',
+    unit: 'milliseconds',
+    min: 0,
+    max: 30000,
+    step: 50,
+    display: (v) => formatMs(v),
+  },
   flushDelayMs: {
     control: 'slider',
     term: 'dirty-write',
@@ -1174,6 +1212,9 @@ const FIELD_GROUPS: { title: string; fields: ReadonlySet<Field> }[] = [
       'intervalMs',
       'batchSize',
       'bulkheadMax',
+      'bulkheadMode',
+      'acquireQueueMax',
+      'acquireTimeoutMs',
       'flushDelayMs',
       'edgeShare',
       'lowPriorityShare',
@@ -1364,6 +1405,41 @@ function NumberRow({
   );
 }
 
+function ChoiceRow({
+  spec,
+  value,
+  locked,
+  onChange,
+}: {
+  spec: ChoiceSpec;
+  value: 'reject' | 'wait';
+  locked?: boolean;
+  onChange: (v: 'reject' | 'wait') => void;
+}) {
+  const id = useId();
+  return (
+    <div className={locked ? 'ins-field-row is-locked' : 'ins-field-row'}>
+      <label className="row-k" htmlFor={id}>
+        {spec.label}
+      </label>
+      <select
+        id={id}
+        className="ins-number"
+        value={value}
+        disabled={locked}
+        onChange={(e) => onChange(e.currentTarget.value as 'reject' | 'wait')}
+      >
+        {spec.options.map((option) => (
+          <option key={option.value} value={option.value}>
+            {option.label}
+          </option>
+        ))}
+      </select>
+      {spec.hint ? <p className="ins-hint">{spec.hint}</p> : null}
+    </div>
+  );
+}
+
 /**
  * One live readout. `tone` is a health class or undefined — at 'ok' the value
  * stays neutral, so a healthy panel carries no colour at all and --warn /
@@ -1512,7 +1588,8 @@ function VitalsMeter({
       const used = stats.bulkheadInFlight ?? 0;
       const fill = limit > 0 ? Math.min(1, used / limit) : 0;
       const rejecting = (stats.bulkheadRejectedRate ?? 0) > 0;
-      const tone = rejecting ? 'is-danger' : toneClass(healthOfLoad(fill));
+      const timingOut = (stats.bulkheadAcquireTimeoutRate ?? 0) > 0;
+      const tone = rejecting || timingOut ? 'is-danger' : toneClass(healthOfLoad(fill));
       return (
         <div className="ins-util">
           <div className="ins-util-head">
@@ -1659,7 +1736,15 @@ function RegionPanel({ stats }: { stats: NodeStats }) {
  * NOTHING breaker-specific while the node was failing fast at 93 rps.
  * A row appears only for the kinds it applies to.
  */
-function KindStatRows({ kind, stats }: { kind: NodeKind; stats: NodeStats }) {
+function KindStatRows({
+  kind,
+  config,
+  stats,
+}: {
+  kind: NodeKind;
+  config: NodeConfig;
+  stats: NodeStats;
+}) {
   switch (kind) {
     case 'cache':
       return (
@@ -1976,6 +2061,29 @@ function KindStatRows({ kind, stats }: { kind: NodeKind; stats: NodeStats }) {
             value={formatRate(stats.bulkheadRejectedRate)}
             tone={(stats.bulkheadRejectedRate ?? 0) > 0 ? 'is-danger' : undefined}
           />
+          {config.bulkheadMode === 'wait' && (
+            <>
+              <StatRow
+                label="Waiting to acquire"
+                value={formatCount(stats.bulkheadWaiting ?? 0)}
+              />
+              <StatRow
+                label="Last acquire wait, measured"
+                value={
+                  stats.bulkheadAcquireLatencyMs === undefined
+                    ? NA
+                    : formatMs(stats.bulkheadAcquireLatencyMs)
+                }
+              />
+              <StatRow
+                label="Acquire timeouts"
+                value={formatRate(stats.bulkheadAcquireTimeoutRate)}
+                tone={
+                  (stats.bulkheadAcquireTimeoutRate ?? 0) > 0 ? 'is-danger' : undefined
+                }
+              />
+            </>
+          )}
         </>
       );
 
@@ -2659,6 +2767,14 @@ function SingleInspector({
                     locked={locked}
                     onChange={(v) => onChange(node.id, { [field]: v })}
                   />
+                ) : spec.control === 'choice' ? (
+                  <ChoiceRow
+                    key={field}
+                    spec={spec}
+                    value={node.config.bulkheadMode ?? 'reject'}
+                    locked={locked}
+                    onChange={(v) => onChange(node.id, { bulkheadMode: v })}
+                  />
                 ) : (
                   <NumberRow
                     key={field}
@@ -2738,7 +2854,7 @@ function SingleInspector({
                   term="offered"
                   value={formatRate(stats.arrivalRate)}
                 />
-                <KindStatRows kind={node.kind} stats={stats} />
+                <KindStatRows kind={node.kind} config={node.config} stats={stats} />
                 <StatRow
                   label="Typical request"
                   term="p50"
@@ -2957,6 +3073,13 @@ function MultiInspector({
                         mixed={mixed}
                         onChange={(v) => applyAll({ [field]: v })}
                       />
+                    ) : spec.control === 'choice' ? (
+                      <ChoiceRow
+                        key={field}
+                        spec={spec}
+                        value={nodes[0]!.config.bulkheadMode ?? 'reject'}
+                        onChange={(v) => applyAll({ bulkheadMode: v })}
+                      />
                     ) : (
                       <NumberRow
                         key={field}

src/components/Metrics.tsx

@@ -150,7 +150,7 @@
 
     if (target > top) {
       lowSince.current = null;
       setTop(target);
       return;
     }
     if (target >= top) {
@@ -706,6 +706,7 @@
   'conn-refused',
   'unauthorized',
   'bulkhead-full',
+  'acquire-timeout',
   'deprioritized',
 ];
 
@@ -723,6 +724,7 @@
   'conn-refused': 'conn refused',
   unauthorized: 'unauthorized',
   'bulkhead-full': 'bulkhead full',
+  'acquire-timeout': 'acquire timeout',
   deprioritized: 'deprioritized',
 };
 
@@ -753,6 +755,7 @@
   'conn-refused': 'conn-refused',
   unauthorized: 'unauthorized',
   'bulkhead-full': 'bulkhead-full',
+  'acquire-timeout': 'bulkhead-full',
   deprioritized: 'deprioritized',
 };
 
@@ -820,7 +823,7 @@
 
   const { top, ticks } = useStickyAxis(max, 1);
 
   const xAt = useMemo(() => {
     const n = samples.length;
     if (n === 0) return () => 0;
     const tEnd = samples[n - 1]!.t;

src/content/glossary.ts

@@ -921,6 +921,15 @@ export const GLOSSARY: GlossaryEntry[] = [
     aliases: ['bulkhead-full', 'pool full'],
     see: ['bulkhead'],
   },
+  {
+    id: 'acquire-timeout',
+    term: 'Acquire timeout',
+    short: 'A request waited too long for a connection slot',
+    why: 'The downstream calls still hold every slot, so this request never got to start one. Shortening downstream work, adding slots, or reducing retry pressure can clear it.',
+    category: 'failure',
+    aliases: ['acquire-timeout'],
+    see: ['bulkhead'],
+  },
   {
     id: 'deprioritized',
     term: 'Deprioritized',

src/share/wire.test.ts

@@ -101,6 +101,9 @@ describe('protocol tables', () => {
       timeoutMs: true,
       retries: true,
       rps: true,
+      bulkheadMode: true,
+      acquireQueueMax: true,
+      acquireTimeoutMs: true,
       traffic: true,
       trafficPeriodS: true,
       targetUtil: true,
@@ -251,6 +254,7 @@ describe('round trip', () => {
       const overrides: Record<string, unknown> = {};
       for (const field of WIRE_FIELDS) {
         if (field === 'traffic') overrides[field] = 'spike';
+        else if (field === 'bulkheadMode') overrides[field] = 'wait';
         else overrides[field] = 7 + i + WIRE_FIELDS.indexOf(field) / 1000;
       }
       return node(`${kind}-1`, kind, 0, 0, overrides as Partial<NodeConfig>);

src/share/wire.ts

@@ -199,6 +199,9 @@ const FIELDS = [
   'intervalMs',
   'batchSize',
   'bulkheadMax',
+  'bulkheadMode',
+  'acquireQueueMax',
+  'acquireTimeoutMs',
   'flushDelayMs',
   'edgeShare',
   'lowPriorityShare',
@@ -287,7 +290,13 @@ const SCHEMA: Record<NodeKind, readonly ConfigField[]> = {
   sidecar: ['outlierAfter', 'openMs', ...COMMON],
   lambda: ['coldStartMs', 'keepWarmMs', 'maxConcurrency', ...COMMON],
   cron: ['intervalMs', 'batchSize', ...COMMON],
-  bulkhead: ['bulkheadMax', ...COMMON],
+  bulkhead: [
+    'bulkheadMax',
+    'bulkheadMode',
+    'acquireQueueMax',
+    'acquireTimeoutMs',
+    ...COMMON,
+  ],
   retryqueue: COMMON,
   transcoder: ['renditions', ...COMMON],
   edgecompute: ['edgeShare', 'cpuMsCap', ...COMMON],
@@ -590,10 +599,12 @@ function overrides(kind: NodeKind, config: NodeConfig): Map<ConfigField, ConfigV
   for (const field of FIELDS) {
     const v = (config as unknown as Record<string, unknown>)[field];
     if (v === undefined || v === def[field]) continue;
-    if (field === 'traffic') {
+    if (field === 'traffic' || field === 'bulkheadMode') {
       if (
         typeof v === 'string' &&
-        (TRAFFIC_PATTERNS as readonly string[]).includes(v)
+        (field === 'traffic'
+          ? (TRAFFIC_PATTERNS as readonly string[]).includes(v)
+          : v === 'reject' || v === 'wait')
       ) {
         out.set(field, v);
       }
@@ -607,6 +618,8 @@ function overrides(kind: NodeKind, config: NodeConfig): Map<ConfigField, ConfigV
 function packValue(w: Writer, field: ConfigField, v: ConfigValue): void {
   if (field === 'traffic') {
     w.varint(TRAFFIC_PATTERNS.indexOf(v as (typeof TRAFFIC_PATTERNS)[number]));
+  } else if (field === 'bulkheadMode') {
+    w.varint(v === 'wait' ? 1 : 0);
   } else {
     w.num(v as number);
   }
@@ -797,6 +810,11 @@ function unpackValue(r: Reader, field: ConfigField): ConfigValue {
     if (t === undefined) throw new WireError('unknown traffic pattern');
     return t;
   }
+  if (field === 'bulkheadMode') {
+    const mode = r.varint();
+    if (mode > 1) throw new WireError('unknown bulkhead mode');
+    return mode === 1 ? 'wait' : 'reject';
+  }
   return r.num();
 }