detached 588ba1c

feat(broker): keep idle remote transports warm with bounded retention (E5)

Replaces the removed client remote.attachment-cache* settings with
broker-owned warm reuse. The pool already kept an idle physical
transport for a fixed 5m; retention is now configurable in broker.json
and bounded by count:

- warmTransports (0-64, default 8): idle transports kept warm; one more
  retires the least recently idled first; 0 retires on release.
- warmIdleTimeout (off or a duration up to 24h, default off): age bound
  on the pool clock. Both defaults match the old client cache.

Returning to a remote opens a new logical stream over the warm,
authenticated SSH/QUIC transport: no bootstrap or handshake. Warm
transports never pin broker idle shutdown; Pool.Close retires them.
The broker logs broker_remote_dial per remote bootstrap, and the
warm-reuse harness scenarios now assert zero dials after the first
remote attach instead of the old suspend/activate events.
commit
588ba1cb72c5711d11d5ee7143ea7acb5897648b
author
brice · · 2026-09-22 20:33 UTC
parent
818dc48
signature
GPG VerifiedCD88F9E564E0102646FB6077A1A07C7C7C3E349D · arch-gpg Verified for the committer email
14 files · +561 -80
M docs/configuration.md +20 -0
@@ -138,6 +138,26 @@ Remote host commands, listing, and successful direct-attach learning are always
138138
139139 `vev ls <host>` and `vev ls --all` run `ssh -- <host> 'vev cmd remote-catalog --json'` for each known host. OpenSSH resolves aliases and connection settings from your SSH config. Remote session names appear as `session@host`. `vev ls --all` prints local sessions first, then remote sessions in merged host order. A catalog failure is reported after the successful output with the host and error; the command exits non-zero so partial output is not mistaken for a complete inventory.
140140
141+### Warm remote transports
142+
143+The connection broker keeps a remote's physical transport warm after the last terminal client leaves it. Returning to that remote opens a new logical stream over the retained, already authenticated SSH or QUIC connection: no second SSH bootstrap, QUIC handshake, or remote helper start. Broker observation probes borrow the same pooled transport instead of dialing their own. The remote daemon sees a fresh attach; session state lives on the daemon, so the screen and scrollback are unchanged.
144+
145+Retention is set in the broker configuration, `~/.config/vev/broker.json` (`$XDG_CONFIG_HOME` respected), which the broker creates on first use. It is read when the broker starts.
146+
147+```json
148+{
149+ "marker": "vev.broker.offline/v1",
150+ "registrations": [],
151+ "warmTransports": 8,
152+ "warmIdleTimeout": "off"
153+}
154+```
155+
156+- `warmTransports`: 0–64 idle transports kept warm, default 8. When one more goes idle, the least recently idled one closes. `0` closes a transport as soon as its last stream ends. Transports with an attached client never count.
157+- `warmIdleTimeout`: `off` (default, no age bound) or a positive Go duration up to `24h`, such as `30s` or `5m`. A warm transport closes that long after it went idle.
158+
159+A remote closing the connection also evicts it; the next visit dials normally. Warm transports never keep the broker alive: when the broker stops after its idle grace, it closes them. Aliases stay distinct unless the broker authenticated them as the same daemon under the same connection policy. These keys replace the client-side `remote.attachment-cache`, `remote.attachment-cache-capacity`, and `remote.attachment-cache-idle-timeout` settings, which no longer exist.
160+
141161 ## Logs and durable state
142162
143163 Set `VEV_LOG=debug`, `VEV_LOG=warn`, or `VEV_LOG=error` to change verbosity; the default is `info`. JSON-line logs such as `vev-daemon.log` live in `$XDG_STATE_HOME/vev`, or `~/.local/state/vev` when unset. The same state directory contains the strict session catalogue, any private pre-migration catalogue backup, notices, `hosts.json`, and `snapshots/`. The lifecycle lock and socket live in `$XDG_RUNTIME_DIR/vev` (with platform runtime fallbacks).
@@ -138,6 +138,26 @@ Remote host commands, listing, and successful direct-attach learning are always
138 138
139 `vev ls <host>` and `vev ls --all` run `ssh -- <host> 'vev cmd remote-catalog --json'` for each known host. OpenSSH resolves aliases and connection settings from your SSH config. Remote session names appear as `session@host`. `vev ls --all` prints local sessions first, then remote sessions in merged host order. A catalog failure is reported after the successful output with the host and error; the command exits non-zero so partial output is not mistaken for a complete inventory.139 `vev ls <host>` and `vev ls --all` run `ssh -- <host> 'vev cmd remote-catalog --json'` for each known host. OpenSSH resolves aliases and connection settings from your SSH config. Remote session names appear as `session@host`. `vev ls --all` prints local sessions first, then remote sessions in merged host order. A catalog failure is reported after the successful output with the host and error; the command exits non-zero so partial output is not mistaken for a complete inventory.
140 140
141+### Warm remote transports
142+
143+The connection broker keeps a remote's physical transport warm after the last terminal client leaves it. Returning to that remote opens a new logical stream over the retained, already authenticated SSH or QUIC connection: no second SSH bootstrap, QUIC handshake, or remote helper start. Broker observation probes borrow the same pooled transport instead of dialing their own. The remote daemon sees a fresh attach; session state lives on the daemon, so the screen and scrollback are unchanged.
144+
145+Retention is set in the broker configuration, `~/.config/vev/broker.json` (`$XDG_CONFIG_HOME` respected), which the broker creates on first use. It is read when the broker starts.
146+
147+```json
148+{
149+ "marker": "vev.broker.offline/v1",
150+ "registrations": [],
151+ "warmTransports": 8,
152+ "warmIdleTimeout": "off"
153+}
154+```
155+
156+- `warmTransports`: 0–64 idle transports kept warm, default 8. When one more goes idle, the least recently idled one closes. `0` closes a transport as soon as its last stream ends. Transports with an attached client never count.
157+- `warmIdleTimeout`: `off` (default, no age bound) or a positive Go duration up to `24h`, such as `30s` or `5m`. A warm transport closes that long after it went idle.
158+
159+A remote closing the connection also evicts it; the next visit dials normally. Warm transports never keep the broker alive: when the broker stops after its idle grace, it closes them. Aliases stay distinct unless the broker authenticated them as the same daemon under the same connection policy. These keys replace the client-side `remote.attachment-cache`, `remote.attachment-cache-capacity`, and `remote.attachment-cache-idle-timeout` settings, which no longer exist.
160+
141 ## Logs and durable state161 ## Logs and durable state
142 162
143 Set `VEV_LOG=debug`, `VEV_LOG=warn`, or `VEV_LOG=error` to change verbosity; the default is `info`. JSON-line logs such as `vev-daemon.log` live in `$XDG_STATE_HOME/vev`, or `~/.local/state/vev` when unset. The same state directory contains the strict session catalogue, any private pre-migration catalogue backup, notices, `hosts.json`, and `snapshots/`. The lifecycle lock and socket live in `$XDG_RUNTIME_DIR/vev` (with platform runtime fallbacks).163 Set `VEV_LOG=debug`, `VEV_LOG=warn`, or `VEV_LOG=error` to change verbosity; the default is `info`. JSON-line logs such as `vev-daemon.log` live in `$XDG_STATE_HOME/vev`, or `~/.local/state/vev` when unset. The same state directory contains the strict session catalogue, any private pre-migration catalogue backup, notices, `hosts.json`, and `snapshots/`. The lifecycle lock and socket live in `$XDG_RUNTIME_DIR/vev` (with platform runtime fallbacks).
M internal/adapters/brokerconfig/config.go +62 -2
@@ -64,6 +64,16 @@ const MaxConfigBytes = 1 << 20
6464 // cannot make an offline broker effectively immortal.
6565 const MaxIdleGrace = 24 * time.Hour
6666
67+// Warm transport retention (the broker-owned replacement for the removed
68+// client remote.attachment-cache keys). DefaultWarmTransports matches that
69+// cache's default capacity; MaxWarmTransports matches the pool's physical
70+// bound. The default idle timeout is off (no age bound), also matching it;
71+// a provisioned timeout is bounded by MaxIdleGrace.
72+const (
73+ DefaultWarmTransports = 8
74+ MaxWarmTransports = 64
75+)
76+
6777 // Sandbox directory names derived from one offline root. SpawnDirName is the
6878 // private directory under Runtime that holds the descriptor-backed
6979 // spawn-election lock, deliberately separate from the broker lifetime lock in
@@ -273,6 +283,10 @@ type Config struct {
273283 // provisioning, use the built-in default" from "provisioned".
274284 idleGrace time.Duration
275285 hasIdleGrace bool
286+ // warmTransports and warmIdleTimeout bound warm physical transport reuse;
287+ // see WarmTransports and WarmIdleTimeout.
288+ warmTransports int
289+ warmIdleTimeout time.Duration
276290 }
277291
278292 // Load reads and strictly validates the config marker inside layout.Root.
@@ -345,6 +359,9 @@ func loadPath(layout Layout, path string, authoritativeLocal *LocalBinding) (*Co
345359 }
346360 config.idleGrace, config.hasIdleGrace = grace, true
347361 }
362+ if config.warmTransports, config.warmIdleTimeout, err = document.warm(); err != nil {
363+ return nil, err
364+ }
348365 pooledRoutes := make(map[pooledIdentity]string, len(document.Registrations)+1)
349366 for i, entry := range document.Registrations {
350367 registration, err := entry.validate()
@@ -496,6 +513,25 @@ func (c *Config) IdleGrace() (time.Duration, bool) {
496513 return c.idleGrace, c.hasIdleGrace
497514 }
498515
516+// WarmTransports returns how many idle physical transports the broker keeps
517+// warm for reuse. Zero disables warm reuse. A nil configuration uses the
518+// default.
519+func (c *Config) WarmTransports() int {
520+ if c == nil {
521+ return DefaultWarmTransports
522+ }
523+ return c.warmTransports
524+}
525+
526+// WarmIdleTimeout returns how long an idle physical transport stays warm.
527+// Zero means no age bound.
528+func (c *Config) WarmIdleTimeout() time.Duration {
529+ if c == nil {
530+ return 0
531+ }
532+ return c.warmIdleTimeout
533+}
534+
499535 // RouteByAddress resolves one opaque pool address back to the route it was
500536 // derived from. The address is produced only by this configuration's own
501537 // routes, so an unknown address is refused rather than dialed.
@@ -545,8 +581,32 @@ type configDocument struct {
545581 // duration string (for example "90s"). It is the single deterministic source
546582 // for the idle grace, because a status probe cannot read a running broker's
547583 // effective grace through the broker IPC protocol.
548- IdleGrace string `json:"idleGrace,omitempty"`
549- Registrations []registrationRaw `json:"registrations"`
584+ IdleGrace string `json:"idleGrace,omitempty"`
585+ // WarmTransports optionally bounds idle physical transports kept warm for
586+ // reuse (0 through MaxWarmTransports; 0 disables warm reuse).
587+ WarmTransports *int `json:"warmTransports,omitempty"`
588+ // WarmIdleTimeout optionally expires a warm transport: "off" (no age
589+ // bound, the default) or a positive Go duration up to MaxIdleGrace.
590+ WarmIdleTimeout string `json:"warmIdleTimeout,omitempty"`
591+ Registrations []registrationRaw `json:"registrations"`
592+}
593+
594+func (d configDocument) warm() (int, time.Duration, error) {
595+ count := DefaultWarmTransports
596+ if d.WarmTransports != nil {
597+ count = *d.WarmTransports
598+ if count < 0 || count > MaxWarmTransports {
599+ return 0, 0, fmt.Errorf("brokerconfig: %s: warmTransports must be between 0 and %d", ConfigFileName, MaxWarmTransports)
600+ }
601+ }
602+ if d.WarmIdleTimeout == "" || d.WarmIdleTimeout == "off" {
603+ return count, 0, nil
604+ }
605+ timeout, err := time.ParseDuration(d.WarmIdleTimeout)
606+ if err != nil || timeout <= 0 || timeout > MaxIdleGrace {
607+ return 0, 0, fmt.Errorf("brokerconfig: %s: warmIdleTimeout %q must be off or a positive duration of at most %s", ConfigFileName, d.WarmIdleTimeout, MaxIdleGrace)
608+ }
609+ return count, timeout, nil
550610 }
551611
552612 // localDocument is the strict on-disk shape of the broker-owned local binding.
@@ -64,6 +64,16 @@ const MaxConfigBytes = 1 << 20
64 // cannot make an offline broker effectively immortal.64 // cannot make an offline broker effectively immortal.
65 const MaxIdleGrace = 24 * time.Hour65 const MaxIdleGrace = 24 * time.Hour
66 66
67+// Warm transport retention (the broker-owned replacement for the removed
68+// client remote.attachment-cache keys). DefaultWarmTransports matches that
69+// cache's default capacity; MaxWarmTransports matches the pool's physical
70+// bound. The default idle timeout is off (no age bound), also matching it;
71+// a provisioned timeout is bounded by MaxIdleGrace.
72+const (
73+ DefaultWarmTransports = 8
74+ MaxWarmTransports = 64
75+)
76+
67 // Sandbox directory names derived from one offline root. SpawnDirName is the77 // Sandbox directory names derived from one offline root. SpawnDirName is the
68 // private directory under Runtime that holds the descriptor-backed78 // private directory under Runtime that holds the descriptor-backed
69 // spawn-election lock, deliberately separate from the broker lifetime lock in79 // spawn-election lock, deliberately separate from the broker lifetime lock in
@@ -273,6 +283,10 @@ type Config struct {
273 // provisioning, use the built-in default" from "provisioned".283 // provisioning, use the built-in default" from "provisioned".
274 idleGrace time.Duration284 idleGrace time.Duration
275 hasIdleGrace bool285 hasIdleGrace bool
286+ // warmTransports and warmIdleTimeout bound warm physical transport reuse;
287+ // see WarmTransports and WarmIdleTimeout.
288+ warmTransports int
289+ warmIdleTimeout time.Duration
276 }290 }
277 291
278 // Load reads and strictly validates the config marker inside layout.Root.292 // Load reads and strictly validates the config marker inside layout.Root.
@@ -345,6 +359,9 @@ func loadPath(layout Layout, path string, authoritativeLocal *LocalBinding) (*Co
345 }359 }
346 config.idleGrace, config.hasIdleGrace = grace, true360 config.idleGrace, config.hasIdleGrace = grace, true
347 }361 }
362+ if config.warmTransports, config.warmIdleTimeout, err = document.warm(); err != nil {
363+ return nil, err
364+ }
348 pooledRoutes := make(map[pooledIdentity]string, len(document.Registrations)+1)365 pooledRoutes := make(map[pooledIdentity]string, len(document.Registrations)+1)
349 for i, entry := range document.Registrations {366 for i, entry := range document.Registrations {
350 registration, err := entry.validate()367 registration, err := entry.validate()
@@ -496,6 +513,25 @@ func (c *Config) IdleGrace() (time.Duration, bool) {
496 return c.idleGrace, c.hasIdleGrace513 return c.idleGrace, c.hasIdleGrace
497 }514 }
498 515
516+// WarmTransports returns how many idle physical transports the broker keeps
517+// warm for reuse. Zero disables warm reuse. A nil configuration uses the
518+// default.
519+func (c *Config) WarmTransports() int {
520+ if c == nil {
521+ return DefaultWarmTransports
522+ }
523+ return c.warmTransports
524+}
525+
526+// WarmIdleTimeout returns how long an idle physical transport stays warm.
527+// Zero means no age bound.
528+func (c *Config) WarmIdleTimeout() time.Duration {
529+ if c == nil {
530+ return 0
531+ }
532+ return c.warmIdleTimeout
533+}
534+
499 // RouteByAddress resolves one opaque pool address back to the route it was535 // RouteByAddress resolves one opaque pool address back to the route it was
500 // derived from. The address is produced only by this configuration's own536 // derived from. The address is produced only by this configuration's own
501 // routes, so an unknown address is refused rather than dialed.537 // routes, so an unknown address is refused rather than dialed.
@@ -545,8 +581,32 @@ type configDocument struct {
545 // duration string (for example "90s"). It is the single deterministic source581 // duration string (for example "90s"). It is the single deterministic source
546 // for the idle grace, because a status probe cannot read a running broker's582 // for the idle grace, because a status probe cannot read a running broker's
547 // effective grace through the broker IPC protocol.583 // effective grace through the broker IPC protocol.
548- IdleGrace string `json:"idleGrace,omitempty"`584+ IdleGrace string `json:"idleGrace,omitempty"`
549- Registrations []registrationRaw `json:"registrations"`585+ // WarmTransports optionally bounds idle physical transports kept warm for
586+ // reuse (0 through MaxWarmTransports; 0 disables warm reuse).
587+ WarmTransports *int `json:"warmTransports,omitempty"`
588+ // WarmIdleTimeout optionally expires a warm transport: "off" (no age
589+ // bound, the default) or a positive Go duration up to MaxIdleGrace.
590+ WarmIdleTimeout string `json:"warmIdleTimeout,omitempty"`
591+ Registrations []registrationRaw `json:"registrations"`
592+}
593+
594+func (d configDocument) warm() (int, time.Duration, error) {
595+ count := DefaultWarmTransports
596+ if d.WarmTransports != nil {
597+ count = *d.WarmTransports
598+ if count < 0 || count > MaxWarmTransports {
599+ return 0, 0, fmt.Errorf("brokerconfig: %s: warmTransports must be between 0 and %d", ConfigFileName, MaxWarmTransports)
600+ }
601+ }
602+ if d.WarmIdleTimeout == "" || d.WarmIdleTimeout == "off" {
603+ return count, 0, nil
604+ }
605+ timeout, err := time.ParseDuration(d.WarmIdleTimeout)
606+ if err != nil || timeout <= 0 || timeout > MaxIdleGrace {
607+ return 0, 0, fmt.Errorf("brokerconfig: %s: warmIdleTimeout %q must be off or a positive duration of at most %s", ConfigFileName, d.WarmIdleTimeout, MaxIdleGrace)
608+ }
609+ return count, timeout, nil
550 }610 }
551 611
552 // localDocument is the strict on-disk shape of the broker-owned local binding.612 // localDocument is the strict on-disk shape of the broker-owned local binding.
M internal/adapters/brokerconfig/config_test.go +48 -0
@@ -656,3 +656,51 @@ func TestLoadIdleGrace(t *testing.T) {
656656 })
657657 }
658658 }
659+
660+func TestLoadWarmTransports(t *testing.T) {
661+ document := func(extra map[string]any) map[string]any {
662+ d := map[string]any{"marker": Marker, "registrations": []any{}}
663+ for k, v := range extra {
664+ d[k] = v
665+ }
666+ return d
667+ }
668+ tests := []struct {
669+ name string
670+ document map[string]any
671+ wantCount int
672+ wantTimeout time.Duration
673+ wantErr string
674+ }{
675+ {name: "defaults match the old client cache", document: document(nil), wantCount: DefaultWarmTransports},
676+ {name: "explicit off", document: document(map[string]any{"warmIdleTimeout": "off"}), wantCount: DefaultWarmTransports},
677+ {name: "bounded count and grace", document: document(map[string]any{"warmTransports": 2, "warmIdleTimeout": "5m"}), wantCount: 2, wantTimeout: 5 * time.Minute},
678+ {name: "zero disables warm reuse", document: document(map[string]any{"warmTransports": 0}), wantCount: 0},
679+ {name: "maximum count", document: document(map[string]any{"warmTransports": MaxWarmTransports}), wantCount: MaxWarmTransports},
680+ {name: "negative count", document: document(map[string]any{"warmTransports": -1}), wantErr: "warmTransports must be between"},
681+ {name: "oversized count", document: document(map[string]any{"warmTransports": MaxWarmTransports + 1}), wantErr: "warmTransports must be between"},
682+ {name: "non-integer count", document: document(map[string]any{"warmTransports": "8"}), wantErr: "warmTransports"},
683+ {name: "invalid timeout", document: document(map[string]any{"warmIdleTimeout": "soon"}), wantErr: "warmIdleTimeout"},
684+ {name: "zero timeout", document: document(map[string]any{"warmIdleTimeout": "0s"}), wantErr: "warmIdleTimeout"},
685+ {name: "oversized timeout", document: document(map[string]any{"warmIdleTimeout": (MaxIdleGrace + time.Second).String()}), wantErr: "warmIdleTimeout"},
686+ }
687+ for _, tt := range tests {
688+ t.Run(tt.name, func(t *testing.T) {
689+ root := t.TempDir()
690+ writeConfig(t, root, tt.document, 0o600)
691+ layout, err := ResolveLayout(root, nil)
692+ require.NoError(t, err)
693+ config, err := Load(layout)
694+ if tt.wantErr != "" {
695+ require.ErrorContains(t, err, tt.wantErr)
696+ return
697+ }
698+ require.NoError(t, err)
699+ require.Equal(t, tt.wantCount, config.WarmTransports())
700+ require.Equal(t, tt.wantTimeout, config.WarmIdleTimeout())
701+ })
702+ }
703+ var none *Config
704+ require.Equal(t, DefaultWarmTransports, none.WarmTransports())
705+ require.Zero(t, none.WarmIdleTimeout())
706+}
@@ -656,3 +656,51 @@ func TestLoadIdleGrace(t *testing.T) {
656 })656 })
657 }657 }
658 }658 }
659+
660+func TestLoadWarmTransports(t *testing.T) {
661+ document := func(extra map[string]any) map[string]any {
662+ d := map[string]any{"marker": Marker, "registrations": []any{}}
663+ for k, v := range extra {
664+ d[k] = v
665+ }
666+ return d
667+ }
668+ tests := []struct {
669+ name string
670+ document map[string]any
671+ wantCount int
672+ wantTimeout time.Duration
673+ wantErr string
674+ }{
675+ {name: "defaults match the old client cache", document: document(nil), wantCount: DefaultWarmTransports},
676+ {name: "explicit off", document: document(map[string]any{"warmIdleTimeout": "off"}), wantCount: DefaultWarmTransports},
677+ {name: "bounded count and grace", document: document(map[string]any{"warmTransports": 2, "warmIdleTimeout": "5m"}), wantCount: 2, wantTimeout: 5 * time.Minute},
678+ {name: "zero disables warm reuse", document: document(map[string]any{"warmTransports": 0}), wantCount: 0},
679+ {name: "maximum count", document: document(map[string]any{"warmTransports": MaxWarmTransports}), wantCount: MaxWarmTransports},
680+ {name: "negative count", document: document(map[string]any{"warmTransports": -1}), wantErr: "warmTransports must be between"},
681+ {name: "oversized count", document: document(map[string]any{"warmTransports": MaxWarmTransports + 1}), wantErr: "warmTransports must be between"},
682+ {name: "non-integer count", document: document(map[string]any{"warmTransports": "8"}), wantErr: "warmTransports"},
683+ {name: "invalid timeout", document: document(map[string]any{"warmIdleTimeout": "soon"}), wantErr: "warmIdleTimeout"},
684+ {name: "zero timeout", document: document(map[string]any{"warmIdleTimeout": "0s"}), wantErr: "warmIdleTimeout"},
685+ {name: "oversized timeout", document: document(map[string]any{"warmIdleTimeout": (MaxIdleGrace + time.Second).String()}), wantErr: "warmIdleTimeout"},
686+ }
687+ for _, tt := range tests {
688+ t.Run(tt.name, func(t *testing.T) {
689+ root := t.TempDir()
690+ writeConfig(t, root, tt.document, 0o600)
691+ layout, err := ResolveLayout(root, nil)
692+ require.NoError(t, err)
693+ config, err := Load(layout)
694+ if tt.wantErr != "" {
695+ require.ErrorContains(t, err, tt.wantErr)
696+ return
697+ }
698+ require.NoError(t, err)
699+ require.Equal(t, tt.wantCount, config.WarmTransports())
700+ require.Equal(t, tt.wantTimeout, config.WarmIdleTimeout())
701+ })
702+ }
703+ var none *Config
704+ require.Equal(t, DefaultWarmTransports, none.WarmTransports())
705+ require.Zero(t, none.WarmIdleTimeout())
706+}
M internal/adapters/daemonmux/pool_integration_test.go +1 -1
@@ -51,7 +51,7 @@ func p3cRemoteRequest(connection ports.BrokerConnectionID, endpoint string, stre
5151
5252 // p3cPoolLimits bounds one pool with room for two aliases and two streams.
5353 func p3cPoolLimits() broker.PoolLimits {
54- return broker.PoolLimits{Physical: 2, Clients: 2, Streams: 4, StreamsPerClient: 4, Idle: time.Hour}
54+ return broker.PoolLimits{Physical: 2, Clients: 2, Streams: 4, StreamsPerClient: 4, Warm: 2, Idle: time.Hour}
5555 }
5656
5757 // TestPoolSharesCompatibleAliasesOverOnePhysical proves the pool pools two
@@ -51,7 +51,7 @@ func p3cRemoteRequest(connection ports.BrokerConnectionID, endpoint string, stre
51 51
52 // p3cPoolLimits bounds one pool with room for two aliases and two streams.52 // p3cPoolLimits bounds one pool with room for two aliases and two streams.
53 func p3cPoolLimits() broker.PoolLimits {53 func p3cPoolLimits() broker.PoolLimits {
54- return broker.PoolLimits{Physical: 2, Clients: 2, Streams: 4, StreamsPerClient: 4, Idle: time.Hour}54+ return broker.PoolLimits{Physical: 2, Clients: 2, Streams: 4, StreamsPerClient: 4, Warm: 2, Idle: time.Hour}
55 }55 }
56 56
57 // TestPoolSharesCompatibleAliasesOverOnePhysical proves the pool pools two57 // TestPoolSharesCompatibleAliasesOverOnePhysical proves the pool pools two
M internal/app/broker_offline.go +11 -2
@@ -59,7 +59,16 @@ var brokerServePoolLimits = broker.PoolLimits{
5959 Clients: brokeripc.DefaultMaxClients,
6060 Streams: 256,
6161 StreamsPerClient: 64,
62- Idle: 5 * time.Minute,
62+ Warm: brokerconfig.DefaultWarmTransports,
63+}
64+
65+// brokerPoolLimits applies the configured warm transport retention to the
66+// fixed pool bounds.
67+func brokerPoolLimits(config *brokerconfig.Config) broker.PoolLimits {
68+ limits := brokerServePoolLimits
69+ limits.Warm = config.WarmTransports()
70+ limits.Idle = config.WarmIdleTimeout()
71+ return limits
6372 }
6473
6574 // brokerServeOptions is the parsed hidden invocation.
@@ -429,7 +438,7 @@ func runBrokerServe(ctx context.Context, options brokerServeOptions, deps broker
429438 if err := supervisor.RegisterRunner("registry", registry); err != nil {
430439 return err
431440 }
432- pool, err := broker.NewPool(epoch, resolver, binder, connector, clk, brokerServePoolLimits)
441+ pool, err := broker.NewPool(epoch, resolver, binder, connector, clk, brokerPoolLimits(config))
433442 if err != nil {
434443 return err
435444 }
@@ -59,7 +59,16 @@ var brokerServePoolLimits = broker.PoolLimits{
59 Clients: brokeripc.DefaultMaxClients,59 Clients: brokeripc.DefaultMaxClients,
60 Streams: 256,60 Streams: 256,
61 StreamsPerClient: 64,61 StreamsPerClient: 64,
62- Idle: 5 * time.Minute,62+ Warm: brokerconfig.DefaultWarmTransports,
63+}
64+
65+// brokerPoolLimits applies the configured warm transport retention to the
66+// fixed pool bounds.
67+func brokerPoolLimits(config *brokerconfig.Config) broker.PoolLimits {
68+ limits := brokerServePoolLimits
69+ limits.Warm = config.WarmTransports()
70+ limits.Idle = config.WarmIdleTimeout()
71+ return limits
63 }72 }
64 73
65 // brokerServeOptions is the parsed hidden invocation.74 // brokerServeOptions is the parsed hidden invocation.
@@ -429,7 +438,7 @@ func runBrokerServe(ctx context.Context, options brokerServeOptions, deps broker
429 if err := supervisor.RegisterRunner("registry", registry); err != nil {438 if err := supervisor.RegisterRunner("registry", registry); err != nil {
430 return err439 return err
431 }440 }
432- pool, err := broker.NewPool(epoch, resolver, binder, connector, clk, brokerServePoolLimits)441+ pool, err := broker.NewPool(epoch, resolver, binder, connector, clk, brokerPoolLimits(config))
433 if err != nil {442 if err != nil {
434 return err443 return err
435 }444 }
M internal/app/broker_offline_mux.go +5 -0
@@ -262,6 +262,11 @@ func brokerRouteAuthorityMatches(config *brokerconfig.Config, target ports.Broke
262262 // through the broker-owned local daemon start; every other Unix route is a
263263 // provisioned endpoint carriage that the broker only ever dials.
264264 func dialBrokerRoute(ctx context.Context, route brokerconfig.Route, target ports.BrokerDialTarget, log *slog.Logger) (daemonmux.RawFramedTransport, error) {
265+ if !target.Fence.Local && log != nil {
266+ // One line per remote physical bootstrap: warm reuse is verified by
267+ // the absence of this event, not by timing.
268+ log.Info("broker_remote_dial", "endpoint", target.Fence.Registration.Endpoint, "route", string(route.Kind()))
269+ }
265270 switch route.Kind() {
266271 case brokerconfig.RouteUnix:
267272 if target.Fence.Local {
@@ -262,6 +262,11 @@ func brokerRouteAuthorityMatches(config *brokerconfig.Config, target ports.Broke
262 // through the broker-owned local daemon start; every other Unix route is a262 // through the broker-owned local daemon start; every other Unix route is a
263 // provisioned endpoint carriage that the broker only ever dials.263 // provisioned endpoint carriage that the broker only ever dials.
264 func dialBrokerRoute(ctx context.Context, route brokerconfig.Route, target ports.BrokerDialTarget, log *slog.Logger) (daemonmux.RawFramedTransport, error) {264 func dialBrokerRoute(ctx context.Context, route brokerconfig.Route, target ports.BrokerDialTarget, log *slog.Logger) (daemonmux.RawFramedTransport, error) {
265+ if !target.Fence.Local && log != nil {
266+ // One line per remote physical bootstrap: warm reuse is verified by
267+ // the absence of this event, not by timing.
268+ log.Info("broker_remote_dial", "endpoint", target.Fence.Registration.Endpoint, "route", string(route.Kind()))
269+ }
265 switch route.Kind() {270 switch route.Kind() {
266 case brokerconfig.RouteUnix:271 case brokerconfig.RouteUnix:
267 if target.Fence.Local {272 if target.Fence.Local {
M internal/app/broker_offline_mux_test.go +15 -5
@@ -574,6 +574,9 @@ func TestBrokerMuxDialFloodedStderrIsReapedAndSanitized(t *testing.T) {
574574 require.NoError(t, waitForProcessExit(pid, 3*time.Second), "the flooding bootstrap must be reaped")
575575
576576 record := decodeBootstrapDiagnostic(t, logs.Bytes())
577+ // Each remote physical bootstrap logs exactly one broker_remote_dial; the
578+ // warm-reuse harness asserts its absence.
579+ require.Equal(t, 1, bytes.Count(logs.Bytes(), []byte(`"msg":"broker_remote_dial"`)))
577580 require.NotEmpty(t, record.Diagnostic)
578581 require.LessOrEqual(t, len(record.Diagnostic), 512, "the captured diagnostic must stay bounded")
579582 for _, r := range record.Diagnostic {
@@ -764,11 +767,18 @@ type bootstrapDiagnostic struct {
764767 }
765768
766769 // decodeBootstrapDiagnostic decodes the single sanitized diagnostic the broker
767-// logs for a failed bootstrap.
770+// logs for a failed bootstrap. Other records (the broker_remote_dial line) are
771+// skipped; exactly one diagnostic must be present.
768772 func decodeBootstrapDiagnostic(t *testing.T, raw []byte) bootstrapDiagnostic {
769773 t.Helper()
770- var record bootstrapDiagnostic
771- require.NoError(t, json.Unmarshal(bytes.TrimSpace(raw), &record))
772- require.Equal(t, "broker_mux_bootstrap_stderr", record.Message)
773- return record
774+ var found []bootstrapDiagnostic
775+ for _, line := range bytes.Split(bytes.TrimSpace(raw), []byte("\n")) {
776+ var record bootstrapDiagnostic
777+ require.NoError(t, json.Unmarshal(line, &record))
778+ if record.Message == "broker_mux_bootstrap_stderr" {
779+ found = append(found, record)
780+ }
781+ }
782+ require.Len(t, found, 1)
783+ return found[0]
774784 }
@@ -574,6 +574,9 @@ func TestBrokerMuxDialFloodedStderrIsReapedAndSanitized(t *testing.T) {
574 require.NoError(t, waitForProcessExit(pid, 3*time.Second), "the flooding bootstrap must be reaped")574 require.NoError(t, waitForProcessExit(pid, 3*time.Second), "the flooding bootstrap must be reaped")
575 575
576 record := decodeBootstrapDiagnostic(t, logs.Bytes())576 record := decodeBootstrapDiagnostic(t, logs.Bytes())
577+ // Each remote physical bootstrap logs exactly one broker_remote_dial; the
578+ // warm-reuse harness asserts its absence.
579+ require.Equal(t, 1, bytes.Count(logs.Bytes(), []byte(`"msg":"broker_remote_dial"`)))
577 require.NotEmpty(t, record.Diagnostic)580 require.NotEmpty(t, record.Diagnostic)
578 require.LessOrEqual(t, len(record.Diagnostic), 512, "the captured diagnostic must stay bounded")581 require.LessOrEqual(t, len(record.Diagnostic), 512, "the captured diagnostic must stay bounded")
579 for _, r := range record.Diagnostic {582 for _, r := range record.Diagnostic {
@@ -764,11 +767,18 @@ type bootstrapDiagnostic struct {
764 }767 }
765 768
766 // decodeBootstrapDiagnostic decodes the single sanitized diagnostic the broker769 // decodeBootstrapDiagnostic decodes the single sanitized diagnostic the broker
767-// logs for a failed bootstrap.770+// logs for a failed bootstrap. Other records (the broker_remote_dial line) are
771+// skipped; exactly one diagnostic must be present.
768 func decodeBootstrapDiagnostic(t *testing.T, raw []byte) bootstrapDiagnostic {772 func decodeBootstrapDiagnostic(t *testing.T, raw []byte) bootstrapDiagnostic {
769 t.Helper()773 t.Helper()
770- var record bootstrapDiagnostic774+ var found []bootstrapDiagnostic
771- require.NoError(t, json.Unmarshal(bytes.TrimSpace(raw), &record))775+ for _, line := range bytes.Split(bytes.TrimSpace(raw), []byte("\n")) {
772- require.Equal(t, "broker_mux_bootstrap_stderr", record.Message)776+ var record bootstrapDiagnostic
773- return record777+ require.NoError(t, json.Unmarshal(line, &record))
778+ if record.Message == "broker_mux_bootstrap_stderr" {
779+ found = append(found, record)
780+ }
781+ }
782+ require.Len(t, found, 1)
783+ return found[0]
774 }784 }
A internal/app/broker_pool_limits_test.go +15 -0
@@ -0,0 +1,15 @@
1+package app
2+
3+import (
4+ "testing"
5+
6+ "github.com/bnema/vev/internal/adapters/brokerconfig"
7+ "github.com/stretchr/testify/require"
8+)
9+
10+func TestBrokerPoolLimitsDefaultWarmRetention(t *testing.T) {
11+ limits := brokerPoolLimits(nil)
12+ require.Equal(t, brokerconfig.DefaultWarmTransports, limits.Warm)
13+ require.Zero(t, limits.Idle, "the default matches the removed client cache: no age bound")
14+ require.LessOrEqual(t, brokerconfig.MaxWarmTransports, limits.Physical)
15+}
@@ -0,0 +1,15 @@
1+package app
2+
3+import (
4+ "testing"
5+
6+ "github.com/bnema/vev/internal/adapters/brokerconfig"
7+ "github.com/stretchr/testify/require"
8+)
9+
10+func TestBrokerPoolLimitsDefaultWarmRetention(t *testing.T) {
11+ limits := brokerPoolLimits(nil)
12+ require.Equal(t, brokerconfig.DefaultWarmTransports, limits.Warm)
13+ require.Zero(t, limits.Idle, "the default matches the removed client cache: no age bound")
14+ require.LessOrEqual(t, brokerconfig.MaxWarmTransports, limits.Physical)
15+}
M internal/usecase/broker/pool.go +70 -9
@@ -13,9 +13,18 @@ import (
1313
1414 // PoolLimits bound live physical keys, clients, reservations (including I/O in
1515 // flight), and logical streams. No unbounded queue or alias cache is retained.
16+//
17+// Warm and Idle bound warm reuse: a published physical transport whose last
18+// logical stream closed stays warm so a returning client opens a new logical
19+// stream without another dial, SSH bootstrap, or QUIC handshake. At most Warm
20+// idle transports are retained; releasing one more retires the least recently
21+// idled first. Zero Warm retires a transport as soon as it goes idle. A
22+// positive Idle retires a transport that long after it went idle; zero Idle
23+// sets no age bound, so retention then ends only through Warm eviction,
24+// physical loss, or Close. Warm transports never pin the broker lifecycle.
1625 type PoolLimits struct {
17- Physical, Clients, Streams, StreamsPerClient int
18- Idle time.Duration
26+ Physical, Clients, Streams, StreamsPerClient, Warm int
27+ Idle time.Duration
1928 }
2029
2130 // poolKey is the exact authenticated (identity, policy) pair the broker pools
@@ -66,6 +75,9 @@ type poolEntry struct {
6675 err error
6776 refs int
6877 retiring bool
78+ // idleOrder is nonzero while the published entry is warm (no reservation)
79+ // and orders warm entries for least-recently-idled eviction.
80+ idleOrder uint64
6981 }
7082
7183 // Pool is transport independent and is not production-composed in P2.2.
@@ -82,6 +94,7 @@ type Pool struct {
8294 cancel context.CancelFunc
8395 closed bool
8496 next uint64
97+ idleOrder uint64
8598 clients map[ports.BrokerConnectionID]*poolClient
8699 entries map[poolKey]*poolEntry
87100 pending map[pendingKey]*poolEntry
@@ -90,7 +103,7 @@ type Pool struct {
90103 }
91104
92105 func NewPool(epoch ports.BrokerEpoch, routes ports.BrokerRouteAuthority, binder ports.BrokerIdentityBinder, connector ports.BrokerEndpointConnector, clock ports.Clock, limits PoolLimits) (*Pool, error) {
93- if epoch == 0 || routes == nil || binder == nil || connector == nil || clock == nil || limits.Physical <= 0 || limits.Clients <= 0 || limits.Streams <= 0 || limits.StreamsPerClient <= 0 || limits.Idle <= 0 {
106+ if epoch == 0 || routes == nil || binder == nil || connector == nil || clock == nil || limits.Physical <= 0 || limits.Clients <= 0 || limits.Streams <= 0 || limits.StreamsPerClient <= 0 || limits.Warm < 0 || limits.Idle < 0 {
94107 return nil, ports.BrokerAdmissionInvalid
95108 }
96109 ctx, cancel := context.WithCancel(context.Background())
@@ -191,12 +204,14 @@ func (p *Pool) OpenStream(ctx context.Context, req ports.BrokerOpenStreamRequest
191204 delete(c.streams, req.Stream)
192205 p.streams--
193206 var abandon context.CancelFunc
207+ var evicted []context.CancelFunc
194208 reservedEntry := r.entry
195209 if reservedEntry != nil {
196210 reservedEntry.refs--
197211 if reservedEntry.refs == 0 {
198212 select {
199213 case <-reservedEntry.ready:
214+ evicted = p.idleLocked(reservedEntry)
200215 default:
201216 reservedEntry.retiring = true
202217 abandon = reservedEntry.cancel
@@ -211,6 +226,9 @@ func (p *Pool) OpenStream(ctx context.Context, req ports.BrokerOpenStreamRequest
211226 if abandon != nil {
212227 abandon()
213228 }
229+ for _, evict := range evicted {
230+ evict()
231+ }
214232 p.wg.Done()
215233 }
216234 success := false
@@ -265,6 +283,7 @@ func (p *Pool) OpenStream(ctx context.Context, req ports.BrokerOpenStreamRequest
265283 go p.runEntry(entry, resolved)
266284 }
267285 entry.refs++
286+ entry.idleOrder = 0
268287 r.entry = entry
269288 p.mu.Unlock()
270289 select {
@@ -340,6 +359,7 @@ func (p *Pool) redirectReservation(entry *poolEntry, reservation *reservation) (
340359 }
341360 entry.refs--
342361 winner.refs++
362+ winner.idleOrder = 0
343363 reservation.entry = winner
344364 return winner, nil
345365 }
@@ -381,15 +401,24 @@ func (p *Pool) runEntry(e *poolEntry, endpoint ports.BrokerDialTarget) {
381401 return
382402 }
383403 if err == nil {
384- timer := p.clock.NewTimer(p.limits.Idle)
385- defer stopPoolTimer(timer)
404+ // Zero Idle sets no age bound: no timer exists and a nil expiry channel
405+ // never fires, so only Warm eviction, physical loss, or Close retire.
406+ var timer ports.Timer
407+ if p.limits.Idle > 0 {
408+ timer = p.clock.NewTimer(p.limits.Idle)
409+ defer stopPoolTimer(timer)
410+ }
386411 for {
387412 p.mu.Lock()
388413 idle := e.refs == 0
389414 p.mu.Unlock()
390- stopPoolTimer(timer)
391- if idle {
392- timer.Reset(p.limits.Idle)
415+ var expiry <-chan time.Time
416+ if timer != nil {
417+ stopPoolTimer(timer)
418+ if idle {
419+ timer.Reset(p.limits.Idle)
420+ expiry = timer.C()
421+ }
393422 }
394423 select {
395424 case <-e.ctx.Done():
@@ -398,7 +427,7 @@ func (p *Pool) runEntry(e *poolEntry, endpoint ports.BrokerDialTarget) {
398427 goto retire
399428 case <-e.wake:
400429 continue
401- case <-timer.C():
430+ case <-expiry:
402431 p.mu.Lock()
403432 if e.refs == 0 {
404433 e.retiring = true
@@ -428,6 +457,38 @@ retire:
428457 p.mu.Unlock()
429458 }
430459
460+// idleLocked marks a published entry whose last reservation just ended as warm
461+// and returns the cancellations that retire warm entries beyond limits.Warm,
462+// least recently idled first. The caller runs them after unlocking. An entry
463+// that failed, lost a redirect race, or is already retiring is never warm.
464+func (p *Pool) idleLocked(e *poolEntry) []context.CancelFunc {
465+ if e.err != nil || e.redirect != nil || e.retiring || p.entries[e.key] != e {
466+ return nil
467+ }
468+ p.idleOrder++
469+ e.idleOrder = p.idleOrder
470+ var warm []*poolEntry
471+ for _, candidate := range p.entries {
472+ if candidate.idleOrder != 0 && candidate.refs == 0 && !candidate.retiring {
473+ warm = append(warm, candidate)
474+ }
475+ }
476+ var evicted []context.CancelFunc
477+ for len(warm) > p.limits.Warm {
478+ oldest := 0
479+ for i := range warm {
480+ if warm[i].idleOrder < warm[oldest].idleOrder {
481+ oldest = i
482+ }
483+ }
484+ victim := warm[oldest]
485+ victim.retiring = true
486+ evicted = append(evicted, victim.cancel)
487+ warm = append(warm[:oldest], warm[oldest+1:]...)
488+ }
489+ return evicted
490+}
491+
431492 // SharedPhysical returns the published physical transport the pool holds for
432493 // one authenticated identity and exact policy, active or warm, without
433494 // reserving it. Observation borrows it to avoid a second bootstrap; the
@@ -13,9 +13,18 @@ import (
13 13
14 // PoolLimits bound live physical keys, clients, reservations (including I/O in14 // PoolLimits bound live physical keys, clients, reservations (including I/O in
15 // flight), and logical streams. No unbounded queue or alias cache is retained.15 // flight), and logical streams. No unbounded queue or alias cache is retained.
16+//
17+// Warm and Idle bound warm reuse: a published physical transport whose last
18+// logical stream closed stays warm so a returning client opens a new logical
19+// stream without another dial, SSH bootstrap, or QUIC handshake. At most Warm
20+// idle transports are retained; releasing one more retires the least recently
21+// idled first. Zero Warm retires a transport as soon as it goes idle. A
22+// positive Idle retires a transport that long after it went idle; zero Idle
23+// sets no age bound, so retention then ends only through Warm eviction,
24+// physical loss, or Close. Warm transports never pin the broker lifecycle.
16 type PoolLimits struct {25 type PoolLimits struct {
17- Physical, Clients, Streams, StreamsPerClient int26+ Physical, Clients, Streams, StreamsPerClient, Warm int
18- Idle time.Duration27+ Idle time.Duration
19 }28 }
20 29
21 // poolKey is the exact authenticated (identity, policy) pair the broker pools30 // poolKey is the exact authenticated (identity, policy) pair the broker pools
@@ -66,6 +75,9 @@ type poolEntry struct {
66 err error75 err error
67 refs int76 refs int
68 retiring bool77 retiring bool
78+ // idleOrder is nonzero while the published entry is warm (no reservation)
79+ // and orders warm entries for least-recently-idled eviction.
80+ idleOrder uint64
69 }81 }
70 82
71 // Pool is transport independent and is not production-composed in P2.2.83 // Pool is transport independent and is not production-composed in P2.2.
@@ -82,6 +94,7 @@ type Pool struct {
82 cancel context.CancelFunc94 cancel context.CancelFunc
83 closed bool95 closed bool
84 next uint6496 next uint64
97+ idleOrder uint64
85 clients map[ports.BrokerConnectionID]*poolClient98 clients map[ports.BrokerConnectionID]*poolClient
86 entries map[poolKey]*poolEntry99 entries map[poolKey]*poolEntry
87 pending map[pendingKey]*poolEntry100 pending map[pendingKey]*poolEntry
@@ -90,7 +103,7 @@ type Pool struct {
90 }103 }
91 104
92 func NewPool(epoch ports.BrokerEpoch, routes ports.BrokerRouteAuthority, binder ports.BrokerIdentityBinder, connector ports.BrokerEndpointConnector, clock ports.Clock, limits PoolLimits) (*Pool, error) {105 func NewPool(epoch ports.BrokerEpoch, routes ports.BrokerRouteAuthority, binder ports.BrokerIdentityBinder, connector ports.BrokerEndpointConnector, clock ports.Clock, limits PoolLimits) (*Pool, error) {
93- if epoch == 0 || routes == nil || binder == nil || connector == nil || clock == nil || limits.Physical <= 0 || limits.Clients <= 0 || limits.Streams <= 0 || limits.StreamsPerClient <= 0 || limits.Idle <= 0 {106+ if epoch == 0 || routes == nil || binder == nil || connector == nil || clock == nil || limits.Physical <= 0 || limits.Clients <= 0 || limits.Streams <= 0 || limits.StreamsPerClient <= 0 || limits.Warm < 0 || limits.Idle < 0 {
94 return nil, ports.BrokerAdmissionInvalid107 return nil, ports.BrokerAdmissionInvalid
95 }108 }
96 ctx, cancel := context.WithCancel(context.Background())109 ctx, cancel := context.WithCancel(context.Background())
@@ -191,12 +204,14 @@ func (p *Pool) OpenStream(ctx context.Context, req ports.BrokerOpenStreamRequest
191 delete(c.streams, req.Stream)204 delete(c.streams, req.Stream)
192 p.streams--205 p.streams--
193 var abandon context.CancelFunc206 var abandon context.CancelFunc
207+ var evicted []context.CancelFunc
194 reservedEntry := r.entry208 reservedEntry := r.entry
195 if reservedEntry != nil {209 if reservedEntry != nil {
196 reservedEntry.refs--210 reservedEntry.refs--
197 if reservedEntry.refs == 0 {211 if reservedEntry.refs == 0 {
198 select {212 select {
199 case <-reservedEntry.ready:213 case <-reservedEntry.ready:
214+ evicted = p.idleLocked(reservedEntry)
200 default:215 default:
201 reservedEntry.retiring = true216 reservedEntry.retiring = true
202 abandon = reservedEntry.cancel217 abandon = reservedEntry.cancel
@@ -211,6 +226,9 @@ func (p *Pool) OpenStream(ctx context.Context, req ports.BrokerOpenStreamRequest
211 if abandon != nil {226 if abandon != nil {
212 abandon()227 abandon()
213 }228 }
229+ for _, evict := range evicted {
230+ evict()
231+ }
214 p.wg.Done()232 p.wg.Done()
215 }233 }
216 success := false234 success := false
@@ -265,6 +283,7 @@ func (p *Pool) OpenStream(ctx context.Context, req ports.BrokerOpenStreamRequest
265 go p.runEntry(entry, resolved)283 go p.runEntry(entry, resolved)
266 }284 }
267 entry.refs++285 entry.refs++
286+ entry.idleOrder = 0
268 r.entry = entry287 r.entry = entry
269 p.mu.Unlock()288 p.mu.Unlock()
270 select {289 select {
@@ -340,6 +359,7 @@ func (p *Pool) redirectReservation(entry *poolEntry, reservation *reservation) (
340 }359 }
341 entry.refs--360 entry.refs--
342 winner.refs++361 winner.refs++
362+ winner.idleOrder = 0
343 reservation.entry = winner363 reservation.entry = winner
344 return winner, nil364 return winner, nil
345 }365 }
@@ -381,15 +401,24 @@ func (p *Pool) runEntry(e *poolEntry, endpoint ports.BrokerDialTarget) {
381 return401 return
382 }402 }
383 if err == nil {403 if err == nil {
384- timer := p.clock.NewTimer(p.limits.Idle)404+ // Zero Idle sets no age bound: no timer exists and a nil expiry channel
385- defer stopPoolTimer(timer)405+ // never fires, so only Warm eviction, physical loss, or Close retire.
406+ var timer ports.Timer
407+ if p.limits.Idle > 0 {
408+ timer = p.clock.NewTimer(p.limits.Idle)
409+ defer stopPoolTimer(timer)
410+ }
386 for {411 for {
387 p.mu.Lock()412 p.mu.Lock()
388 idle := e.refs == 0413 idle := e.refs == 0
389 p.mu.Unlock()414 p.mu.Unlock()
390- stopPoolTimer(timer)415+ var expiry <-chan time.Time
391- if idle {416+ if timer != nil {
392- timer.Reset(p.limits.Idle)417+ stopPoolTimer(timer)
418+ if idle {
419+ timer.Reset(p.limits.Idle)
420+ expiry = timer.C()
421+ }
393 }422 }
394 select {423 select {
395 case <-e.ctx.Done():424 case <-e.ctx.Done():
@@ -398,7 +427,7 @@ func (p *Pool) runEntry(e *poolEntry, endpoint ports.BrokerDialTarget) {
398 goto retire427 goto retire
399 case <-e.wake:428 case <-e.wake:
400 continue429 continue
401- case <-timer.C():430+ case <-expiry:
402 p.mu.Lock()431 p.mu.Lock()
403 if e.refs == 0 {432 if e.refs == 0 {
404 e.retiring = true433 e.retiring = true
@@ -428,6 +457,38 @@ retire:
428 p.mu.Unlock()457 p.mu.Unlock()
429 }458 }
430 459
460+// idleLocked marks a published entry whose last reservation just ended as warm
461+// and returns the cancellations that retire warm entries beyond limits.Warm,
462+// least recently idled first. The caller runs them after unlocking. An entry
463+// that failed, lost a redirect race, or is already retiring is never warm.
464+func (p *Pool) idleLocked(e *poolEntry) []context.CancelFunc {
465+ if e.err != nil || e.redirect != nil || e.retiring || p.entries[e.key] != e {
466+ return nil
467+ }
468+ p.idleOrder++
469+ e.idleOrder = p.idleOrder
470+ var warm []*poolEntry
471+ for _, candidate := range p.entries {
472+ if candidate.idleOrder != 0 && candidate.refs == 0 && !candidate.retiring {
473+ warm = append(warm, candidate)
474+ }
475+ }
476+ var evicted []context.CancelFunc
477+ for len(warm) > p.limits.Warm {
478+ oldest := 0
479+ for i := range warm {
480+ if warm[i].idleOrder < warm[oldest].idleOrder {
481+ oldest = i
482+ }
483+ }
484+ victim := warm[oldest]
485+ victim.retiring = true
486+ evicted = append(evicted, victim.cancel)
487+ warm = append(warm[:oldest], warm[oldest+1:]...)
488+ }
489+ return evicted
490+}
491+
431 // SharedPhysical returns the published physical transport the pool holds for492 // SharedPhysical returns the published physical transport the pool holds for
432 // one authenticated identity and exact policy, active or warm, without493 // one authenticated identity and exact policy, active or warm, without
433 // reserving it. Observation borrows it to avoid a second bootstrap; the494 // reserving it. Observation borrows it to avoid a second bootstrap; the
M internal/usecase/broker/pool_test.go +2 -2
@@ -90,7 +90,7 @@ func setupPool(t testing.TB, connector poolConnector) (*Pool, *manualClock) {
9090 binder := poolBinder(func(_ context.Context, request ports.BrokerIdentityBindingRequest) (ports.BrokerDaemonIdentity, error) {
9191 return request.Identity, nil
9292 })
93- p, err := NewPool(1, resolver, binder, connector, clock, PoolLimits{Physical: 2, Clients: 128, Streams: 128, StreamsPerClient: 128, Idle: time.Minute})
93+ p, err := NewPool(1, resolver, binder, connector, clock, PoolLimits{Physical: 2, Clients: 128, Streams: 128, StreamsPerClient: 128, Warm: 2, Idle: time.Minute})
9494 require.NoError(t, err)
9595 t.Cleanup(func() { require.NoError(t, p.Close()) })
9696 return p, clock
@@ -120,7 +120,7 @@ func TestPoolRefusedResolutionNeverDials(t *testing.T) {
120120 binder := poolBinder(func(_ context.Context, request ports.BrokerIdentityBindingRequest) (ports.BrokerDaemonIdentity, error) {
121121 return request.Identity, nil
122122 })
123- pool, err := NewPool(1, resolver, binder, connector, newManualClock(time.Unix(0, 0)), PoolLimits{Physical: 1, Clients: 8, Streams: 8, StreamsPerClient: 8, Idle: time.Minute})
123+ pool, err := NewPool(1, resolver, binder, connector, newManualClock(time.Unix(0, 0)), PoolLimits{Physical: 1, Clients: 8, Streams: 8, StreamsPerClient: 8, Warm: 1, Idle: time.Minute})
124124 require.NoError(t, err)
125125 t.Cleanup(func() { require.NoError(t, pool.Close()) })
126126
@@ -90,7 +90,7 @@ func setupPool(t testing.TB, connector poolConnector) (*Pool, *manualClock) {
90 binder := poolBinder(func(_ context.Context, request ports.BrokerIdentityBindingRequest) (ports.BrokerDaemonIdentity, error) {90 binder := poolBinder(func(_ context.Context, request ports.BrokerIdentityBindingRequest) (ports.BrokerDaemonIdentity, error) {
91 return request.Identity, nil91 return request.Identity, nil
92 })92 })
93- p, err := NewPool(1, resolver, binder, connector, clock, PoolLimits{Physical: 2, Clients: 128, Streams: 128, StreamsPerClient: 128, Idle: time.Minute})93+ p, err := NewPool(1, resolver, binder, connector, clock, PoolLimits{Physical: 2, Clients: 128, Streams: 128, StreamsPerClient: 128, Warm: 2, Idle: time.Minute})
94 require.NoError(t, err)94 require.NoError(t, err)
95 t.Cleanup(func() { require.NoError(t, p.Close()) })95 t.Cleanup(func() { require.NoError(t, p.Close()) })
96 return p, clock96 return p, clock
@@ -120,7 +120,7 @@ func TestPoolRefusedResolutionNeverDials(t *testing.T) {
120 binder := poolBinder(func(_ context.Context, request ports.BrokerIdentityBindingRequest) (ports.BrokerDaemonIdentity, error) {120 binder := poolBinder(func(_ context.Context, request ports.BrokerIdentityBindingRequest) (ports.BrokerDaemonIdentity, error) {
121 return request.Identity, nil121 return request.Identity, nil
122 })122 })
123- pool, err := NewPool(1, resolver, binder, connector, newManualClock(time.Unix(0, 0)), PoolLimits{Physical: 1, Clients: 8, Streams: 8, StreamsPerClient: 8, Idle: time.Minute})123+ pool, err := NewPool(1, resolver, binder, connector, newManualClock(time.Unix(0, 0)), PoolLimits{Physical: 1, Clients: 8, Streams: 8, StreamsPerClient: 8, Warm: 1, Idle: time.Minute})
124 require.NoError(t, err)124 require.NoError(t, err)
125 t.Cleanup(func() { require.NoError(t, pool.Close()) })125 t.Cleanup(func() { require.NoError(t, pool.Close()) })
126 126
A internal/usecase/broker/pool_warm_test.go +261 -0
@@ -0,0 +1,261 @@
1+package broker
2+
3+import (
4+ "context"
5+ "sync"
6+ "testing"
7+ "time"
8+
9+ "github.com/bnema/vev/internal/ports"
10+ portsmocks "github.com/bnema/vev/internal/ports/mocks"
11+ "github.com/stretchr/testify/mock"
12+ "github.com/stretchr/testify/require"
13+)
14+
15+// warmHarness records every physical the pool dials per identity. The pool
16+// ports are mocks; fakePhysical stays a stateful fake because its Done channel
17+// models transport lifetime across pool workers.
18+type warmHarness struct {
19+ pool *Pool
20+ clock *manualClock
21+ client ports.BrokerConnectionID
22+ mu sync.Mutex
23+ target ports.BrokerDaemonIdentity
24+ dials map[ports.BrokerDaemonIdentity][]*fakePhysical
25+ stream uint64
26+}
27+
28+func newWarmHarness(t *testing.T, warm int, idle time.Duration) *warmHarness {
29+ t.Helper()
30+ h := &warmHarness{clock: newManualClock(time.Unix(0, 0)), target: "alpha", dials: make(map[ports.BrokerDaemonIdentity][]*fakePhysical)}
31+ routes := portsmocks.NewMockBrokerRouteAuthority(t)
32+ routes.EXPECT().ResolveDialTarget(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, r ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
33+ h.mu.Lock()
34+ defer h.mu.Unlock()
35+ return ports.BrokerDialTarget{Fence: ports.BrokerEndpointFence{Local: true}, Policy: r.Policy, Address: "fake", StartMode: r.StartMode, ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: h.target, Bound: true}}, nil
36+ }).Maybe()
37+ binder := portsmocks.NewMockBrokerIdentityBinder(t)
38+ binder.EXPECT().BindAuthenticatedIdentity(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, r ports.BrokerIdentityBindingRequest) (ports.BrokerDaemonIdentity, error) {
39+ return r.Identity, nil
40+ }).Maybe()
41+ connector := portsmocks.NewMockBrokerEndpointConnector(t)
42+ connector.EXPECT().Connect(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
43+ f := &fakePhysical{endpoint: e, done: make(chan struct{})}
44+ h.mu.Lock()
45+ h.dials[e.ExpectedIdentity.Identity] = append(h.dials[e.ExpectedIdentity.Identity], f)
46+ h.mu.Unlock()
47+ return f, nil
48+ }).Maybe()
49+ pool, err := NewPool(1, routes, binder, connector, h.clock, PoolLimits{Physical: 4, Clients: 4, Streams: 16, StreamsPerClient: 16, Warm: warm, Idle: idle})
50+ require.NoError(t, err)
51+ h.pool = pool
52+ t.Cleanup(func() { require.NoError(t, pool.Close()) })
53+ h.client, err = pool.RegisterClient()
54+ require.NoError(t, err)
55+ return h
56+}
57+
58+// visit opens and closes one logical stream to identity, as a client that
59+// attaches to a remote and then navigates away does.
60+func (h *warmHarness) visit(t *testing.T, identity ports.BrokerDaemonIdentity) {
61+ t.Helper()
62+ h.mu.Lock()
63+ h.target = identity
64+ h.stream++
65+ stream := h.stream
66+ h.mu.Unlock()
67+ s, err := h.pool.OpenStream(context.Background(), poolRequest(h.client, stream))
68+ require.NoError(t, err)
69+ require.NoError(t, s.Close())
70+}
71+
72+func (h *warmHarness) physicals(identity ports.BrokerDaemonIdentity) []*fakePhysical {
73+ h.mu.Lock()
74+ defer h.mu.Unlock()
75+ return append([]*fakePhysical(nil), h.dials[identity]...)
76+}
77+
78+func closed(f *fakePhysical) bool {
79+ select {
80+ case <-f.done:
81+ return true
82+ default:
83+ return false
84+ }
85+}
86+
87+// reap advances the fake clock one grace at a time until f retires. Repeating
88+// the advance tolerates a worker that re-arms after a late wake; each step is
89+// a bounded fake-clock move, never a sleep.
90+func (h *warmHarness) reap(t *testing.T, f *fakePhysical, grace time.Duration) {
91+ t.Helper()
92+ require.Eventually(t, func() bool { h.clock.Advance(grace); return closed(f) }, 3*time.Second, time.Millisecond)
93+}
94+
95+func TestPoolWarmReuse(t *testing.T) {
96+ const grace = 5 * time.Minute
97+ tests := []struct {
98+ name string
99+ warm int
100+ idle time.Duration
101+ // run drives local→remote→local→remote style visits and returns the
102+ // dial count expected per identity afterwards.
103+ run func(t *testing.T, h *warmHarness)
104+ want map[ports.BrokerDaemonIdentity]int
105+ }{
106+ {
107+ name: "reuse within grace",
108+ warm: 2, idle: grace,
109+ run: func(t *testing.T, h *warmHarness) {
110+ h.visit(t, "alpha")
111+ h.clock.Advance(grace / 2)
112+ h.visit(t, "alpha")
113+ require.False(t, closed(h.physicals("alpha")[0]))
114+ },
115+ want: map[ports.BrokerDaemonIdentity]int{"alpha": 1},
116+ },
117+ {
118+ name: "reap after grace",
119+ warm: 2, idle: grace,
120+ run: func(t *testing.T, h *warmHarness) {
121+ h.visit(t, "alpha")
122+ h.reap(t, h.physicals("alpha")[0], grace)
123+ require.Eventually(t, func() bool { _, ok := h.pool.SharedPhysical("alpha", poolPolicy()); return !ok }, 3*time.Second, time.Millisecond)
124+ require.Eventually(t, func() bool { h.pool.mu.Lock(); defer h.pool.mu.Unlock(); return len(h.pool.entries) == 0 }, 3*time.Second, time.Millisecond)
125+ h.visit(t, "alpha")
126+ },
127+ want: map[ports.BrokerDaemonIdentity]int{"alpha": 2},
128+ },
129+ {
130+ name: "no age bound keeps warm without timers",
131+ warm: 2, idle: 0,
132+ run: func(t *testing.T, h *warmHarness) {
133+ h.visit(t, "alpha")
134+ h.clock.Advance(24 * time.Hour)
135+ h.visit(t, "alpha")
136+ h.clock.mu.Lock()
137+ timers := len(h.clock.timers)
138+ h.clock.mu.Unlock()
139+ require.Zero(t, timers)
140+ },
141+ want: map[ports.BrokerDaemonIdentity]int{"alpha": 1},
142+ },
143+ {
144+ name: "bound evicts least recently idled",
145+ warm: 1, idle: 0,
146+ run: func(t *testing.T, h *warmHarness) {
147+ h.visit(t, "alpha")
148+ h.visit(t, "beta")
149+ await(t, h.physicals("alpha")[0].done)
150+ require.False(t, closed(h.physicals("beta")[0]))
151+ h.visit(t, "beta")
152+ },
153+ want: map[ports.BrokerDaemonIdentity]int{"alpha": 1, "beta": 1},
154+ },
155+ {
156+ name: "active streams do not count toward the bound",
157+ warm: 1, idle: 0,
158+ run: func(t *testing.T, h *warmHarness) {
159+ h.visit(t, "alpha")
160+ h.mu.Lock()
161+ h.target = "beta"
162+ h.stream++
163+ stream := h.stream
164+ h.mu.Unlock()
165+ held, err := h.pool.OpenStream(context.Background(), poolRequest(h.client, stream))
166+ require.NoError(t, err)
167+ h.visit(t, "alpha")
168+ require.False(t, closed(h.physicals("alpha")[0]))
169+ require.NoError(t, held.Close())
170+ await(t, h.physicals("alpha")[0].done)
171+ },
172+ want: map[ports.BrokerDaemonIdentity]int{"alpha": 1, "beta": 1},
173+ },
174+ {
175+ name: "zero warm retires on release",
176+ warm: 0, idle: 0,
177+ run: func(t *testing.T, h *warmHarness) {
178+ h.visit(t, "alpha")
179+ await(t, h.physicals("alpha")[0].done)
180+ require.Eventually(t, func() bool { h.pool.mu.Lock(); defer h.pool.mu.Unlock(); return len(h.pool.entries) == 0 }, 3*time.Second, time.Millisecond)
181+ h.visit(t, "alpha")
182+ },
183+ want: map[ports.BrokerDaemonIdentity]int{"alpha": 2},
184+ },
185+ }
186+ for _, tt := range tests {
187+ t.Run(tt.name, func(t *testing.T) {
188+ h := newWarmHarness(t, tt.warm, tt.idle)
189+ tt.run(t, h)
190+ for identity, want := range tt.want {
191+ require.Len(t, h.physicals(identity), want, "dials for %s", identity)
192+ }
193+ })
194+ }
195+}
196+
197+func TestPoolCloseRetiresWarmTransports(t *testing.T) {
198+ h := newWarmHarness(t, 2, 0)
199+ h.visit(t, "alpha")
200+ h.visit(t, "beta")
201+ require.NoError(t, h.pool.Close())
202+ for _, identity := range []ports.BrokerDaemonIdentity{"alpha", "beta"} {
203+ require.True(t, closed(h.physicals(identity)[0]), identity)
204+ }
205+ _, ok := h.pool.SharedPhysical("alpha", poolPolicy())
206+ require.False(t, ok)
207+}
208+
209+func TestPoolSharedPhysical(t *testing.T) {
210+ tests := []struct {
211+ name string
212+ identity ports.BrokerDaemonIdentity
213+ policy func() ports.BrokerPolicy
214+ want bool
215+ }{
216+ {name: "warm exact key", identity: "alpha", policy: poolPolicy, want: true},
217+ {name: "unknown identity", identity: "beta", policy: poolPolicy},
218+ {name: "different policy", identity: "alpha", policy: func() ports.BrokerPolicy { p := poolPolicy(); p.Trust = "other"; return p }},
219+ }
220+ for _, tt := range tests {
221+ t.Run(tt.name, func(t *testing.T) {
222+ h := newWarmHarness(t, 1, 0)
223+ h.visit(t, "alpha")
224+ physical, ok := h.pool.SharedPhysical(tt.identity, tt.policy())
225+ require.Equal(t, tt.want, ok)
226+ if tt.want {
227+ require.Same(t, h.physicals("alpha")[0], physical)
228+ // Borrowing neither dials nor refreshes warm order: beta still
229+ // evicts alpha.
230+ h.visit(t, "beta")
231+ await(t, h.physicals("alpha")[0].done)
232+ require.Len(t, h.physicals("alpha"), 1)
233+ }
234+ })
235+ }
236+}
237+
238+func TestNewPoolWarmLimits(t *testing.T) {
239+ tests := []struct {
240+ name string
241+ warm int
242+ idle time.Duration
243+ wantErr bool
244+ }{
245+ {name: "defaults off", warm: 0, idle: 0},
246+ {name: "bounded with grace", warm: 8, idle: time.Minute},
247+ {name: "negative warm", warm: -1, wantErr: true},
248+ {name: "negative idle", warm: 1, idle: -time.Second, wantErr: true},
249+ }
250+ for _, tt := range tests {
251+ t.Run(tt.name, func(t *testing.T) {
252+ pool, err := NewPool(1, portsmocks.NewMockBrokerRouteAuthority(t), portsmocks.NewMockBrokerIdentityBinder(t), portsmocks.NewMockBrokerEndpointConnector(t), newManualClock(time.Unix(0, 0)), PoolLimits{Physical: 1, Clients: 1, Streams: 1, StreamsPerClient: 1, Warm: tt.warm, Idle: tt.idle})
253+ if tt.wantErr {
254+ require.ErrorIs(t, err, ports.BrokerAdmissionInvalid)
255+ return
256+ }
257+ require.NoError(t, err)
258+ require.NoError(t, pool.Close())
259+ })
260+ }
261+}
@@ -0,0 +1,261 @@
1+package broker
2+
3+import (
4+ "context"
5+ "sync"
6+ "testing"
7+ "time"
8+
9+ "github.com/bnema/vev/internal/ports"
10+ portsmocks "github.com/bnema/vev/internal/ports/mocks"
11+ "github.com/stretchr/testify/mock"
12+ "github.com/stretchr/testify/require"
13+)
14+
15+// warmHarness records every physical the pool dials per identity. The pool
16+// ports are mocks; fakePhysical stays a stateful fake because its Done channel
17+// models transport lifetime across pool workers.
18+type warmHarness struct {
19+ pool *Pool
20+ clock *manualClock
21+ client ports.BrokerConnectionID
22+ mu sync.Mutex
23+ target ports.BrokerDaemonIdentity
24+ dials map[ports.BrokerDaemonIdentity][]*fakePhysical
25+ stream uint64
26+}
27+
28+func newWarmHarness(t *testing.T, warm int, idle time.Duration) *warmHarness {
29+ t.Helper()
30+ h := &warmHarness{clock: newManualClock(time.Unix(0, 0)), target: "alpha", dials: make(map[ports.BrokerDaemonIdentity][]*fakePhysical)}
31+ routes := portsmocks.NewMockBrokerRouteAuthority(t)
32+ routes.EXPECT().ResolveDialTarget(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, r ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
33+ h.mu.Lock()
34+ defer h.mu.Unlock()
35+ return ports.BrokerDialTarget{Fence: ports.BrokerEndpointFence{Local: true}, Policy: r.Policy, Address: "fake", StartMode: r.StartMode, ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: h.target, Bound: true}}, nil
36+ }).Maybe()
37+ binder := portsmocks.NewMockBrokerIdentityBinder(t)
38+ binder.EXPECT().BindAuthenticatedIdentity(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, r ports.BrokerIdentityBindingRequest) (ports.BrokerDaemonIdentity, error) {
39+ return r.Identity, nil
40+ }).Maybe()
41+ connector := portsmocks.NewMockBrokerEndpointConnector(t)
42+ connector.EXPECT().Connect(mock.Anything, mock.Anything).RunAndReturn(func(_ context.Context, e ports.BrokerDialTarget) (ports.BrokerPhysicalConnection, error) {
43+ f := &fakePhysical{endpoint: e, done: make(chan struct{})}
44+ h.mu.Lock()
45+ h.dials[e.ExpectedIdentity.Identity] = append(h.dials[e.ExpectedIdentity.Identity], f)
46+ h.mu.Unlock()
47+ return f, nil
48+ }).Maybe()
49+ pool, err := NewPool(1, routes, binder, connector, h.clock, PoolLimits{Physical: 4, Clients: 4, Streams: 16, StreamsPerClient: 16, Warm: warm, Idle: idle})
50+ require.NoError(t, err)
51+ h.pool = pool
52+ t.Cleanup(func() { require.NoError(t, pool.Close()) })
53+ h.client, err = pool.RegisterClient()
54+ require.NoError(t, err)
55+ return h
56+}
57+
58+// visit opens and closes one logical stream to identity, as a client that
59+// attaches to a remote and then navigates away does.
60+func (h *warmHarness) visit(t *testing.T, identity ports.BrokerDaemonIdentity) {
61+ t.Helper()
62+ h.mu.Lock()
63+ h.target = identity
64+ h.stream++
65+ stream := h.stream
66+ h.mu.Unlock()
67+ s, err := h.pool.OpenStream(context.Background(), poolRequest(h.client, stream))
68+ require.NoError(t, err)
69+ require.NoError(t, s.Close())
70+}
71+
72+func (h *warmHarness) physicals(identity ports.BrokerDaemonIdentity) []*fakePhysical {
73+ h.mu.Lock()
74+ defer h.mu.Unlock()
75+ return append([]*fakePhysical(nil), h.dials[identity]...)
76+}
77+
78+func closed(f *fakePhysical) bool {
79+ select {
80+ case <-f.done:
81+ return true
82+ default:
83+ return false
84+ }
85+}
86+
87+// reap advances the fake clock one grace at a time until f retires. Repeating
88+// the advance tolerates a worker that re-arms after a late wake; each step is
89+// a bounded fake-clock move, never a sleep.
90+func (h *warmHarness) reap(t *testing.T, f *fakePhysical, grace time.Duration) {
91+ t.Helper()
92+ require.Eventually(t, func() bool { h.clock.Advance(grace); return closed(f) }, 3*time.Second, time.Millisecond)
93+}
94+
95+func TestPoolWarmReuse(t *testing.T) {
96+ const grace = 5 * time.Minute
97+ tests := []struct {
98+ name string
99+ warm int
100+ idle time.Duration
101+ // run drives local→remote→local→remote style visits and returns the
102+ // dial count expected per identity afterwards.
103+ run func(t *testing.T, h *warmHarness)
104+ want map[ports.BrokerDaemonIdentity]int
105+ }{
106+ {
107+ name: "reuse within grace",
108+ warm: 2, idle: grace,
109+ run: func(t *testing.T, h *warmHarness) {
110+ h.visit(t, "alpha")
111+ h.clock.Advance(grace / 2)
112+ h.visit(t, "alpha")
113+ require.False(t, closed(h.physicals("alpha")[0]))
114+ },
115+ want: map[ports.BrokerDaemonIdentity]int{"alpha": 1},
116+ },
117+ {
118+ name: "reap after grace",
119+ warm: 2, idle: grace,
120+ run: func(t *testing.T, h *warmHarness) {
121+ h.visit(t, "alpha")
122+ h.reap(t, h.physicals("alpha")[0], grace)
123+ require.Eventually(t, func() bool { _, ok := h.pool.SharedPhysical("alpha", poolPolicy()); return !ok }, 3*time.Second, time.Millisecond)
124+ require.Eventually(t, func() bool { h.pool.mu.Lock(); defer h.pool.mu.Unlock(); return len(h.pool.entries) == 0 }, 3*time.Second, time.Millisecond)
125+ h.visit(t, "alpha")
126+ },
127+ want: map[ports.BrokerDaemonIdentity]int{"alpha": 2},
128+ },
129+ {
130+ name: "no age bound keeps warm without timers",
131+ warm: 2, idle: 0,
132+ run: func(t *testing.T, h *warmHarness) {
133+ h.visit(t, "alpha")
134+ h.clock.Advance(24 * time.Hour)
135+ h.visit(t, "alpha")
136+ h.clock.mu.Lock()
137+ timers := len(h.clock.timers)
138+ h.clock.mu.Unlock()
139+ require.Zero(t, timers)
140+ },
141+ want: map[ports.BrokerDaemonIdentity]int{"alpha": 1},
142+ },
143+ {
144+ name: "bound evicts least recently idled",
145+ warm: 1, idle: 0,
146+ run: func(t *testing.T, h *warmHarness) {
147+ h.visit(t, "alpha")
148+ h.visit(t, "beta")
149+ await(t, h.physicals("alpha")[0].done)
150+ require.False(t, closed(h.physicals("beta")[0]))
151+ h.visit(t, "beta")
152+ },
153+ want: map[ports.BrokerDaemonIdentity]int{"alpha": 1, "beta": 1},
154+ },
155+ {
156+ name: "active streams do not count toward the bound",
157+ warm: 1, idle: 0,
158+ run: func(t *testing.T, h *warmHarness) {
159+ h.visit(t, "alpha")
160+ h.mu.Lock()
161+ h.target = "beta"
162+ h.stream++
163+ stream := h.stream
164+ h.mu.Unlock()
165+ held, err := h.pool.OpenStream(context.Background(), poolRequest(h.client, stream))
166+ require.NoError(t, err)
167+ h.visit(t, "alpha")
168+ require.False(t, closed(h.physicals("alpha")[0]))
169+ require.NoError(t, held.Close())
170+ await(t, h.physicals("alpha")[0].done)
171+ },
172+ want: map[ports.BrokerDaemonIdentity]int{"alpha": 1, "beta": 1},
173+ },
174+ {
175+ name: "zero warm retires on release",
176+ warm: 0, idle: 0,
177+ run: func(t *testing.T, h *warmHarness) {
178+ h.visit(t, "alpha")
179+ await(t, h.physicals("alpha")[0].done)
180+ require.Eventually(t, func() bool { h.pool.mu.Lock(); defer h.pool.mu.Unlock(); return len(h.pool.entries) == 0 }, 3*time.Second, time.Millisecond)
181+ h.visit(t, "alpha")
182+ },
183+ want: map[ports.BrokerDaemonIdentity]int{"alpha": 2},
184+ },
185+ }
186+ for _, tt := range tests {
187+ t.Run(tt.name, func(t *testing.T) {
188+ h := newWarmHarness(t, tt.warm, tt.idle)
189+ tt.run(t, h)
190+ for identity, want := range tt.want {
191+ require.Len(t, h.physicals(identity), want, "dials for %s", identity)
192+ }
193+ })
194+ }
195+}
196+
197+func TestPoolCloseRetiresWarmTransports(t *testing.T) {
198+ h := newWarmHarness(t, 2, 0)
199+ h.visit(t, "alpha")
200+ h.visit(t, "beta")
201+ require.NoError(t, h.pool.Close())
202+ for _, identity := range []ports.BrokerDaemonIdentity{"alpha", "beta"} {
203+ require.True(t, closed(h.physicals(identity)[0]), identity)
204+ }
205+ _, ok := h.pool.SharedPhysical("alpha", poolPolicy())
206+ require.False(t, ok)
207+}
208+
209+func TestPoolSharedPhysical(t *testing.T) {
210+ tests := []struct {
211+ name string
212+ identity ports.BrokerDaemonIdentity
213+ policy func() ports.BrokerPolicy
214+ want bool
215+ }{
216+ {name: "warm exact key", identity: "alpha", policy: poolPolicy, want: true},
217+ {name: "unknown identity", identity: "beta", policy: poolPolicy},
218+ {name: "different policy", identity: "alpha", policy: func() ports.BrokerPolicy { p := poolPolicy(); p.Trust = "other"; return p }},
219+ }
220+ for _, tt := range tests {
221+ t.Run(tt.name, func(t *testing.T) {
222+ h := newWarmHarness(t, 1, 0)
223+ h.visit(t, "alpha")
224+ physical, ok := h.pool.SharedPhysical(tt.identity, tt.policy())
225+ require.Equal(t, tt.want, ok)
226+ if tt.want {
227+ require.Same(t, h.physicals("alpha")[0], physical)
228+ // Borrowing neither dials nor refreshes warm order: beta still
229+ // evicts alpha.
230+ h.visit(t, "beta")
231+ await(t, h.physicals("alpha")[0].done)
232+ require.Len(t, h.physicals("alpha"), 1)
233+ }
234+ })
235+ }
236+}
237+
238+func TestNewPoolWarmLimits(t *testing.T) {
239+ tests := []struct {
240+ name string
241+ warm int
242+ idle time.Duration
243+ wantErr bool
244+ }{
245+ {name: "defaults off", warm: 0, idle: 0},
246+ {name: "bounded with grace", warm: 8, idle: time.Minute},
247+ {name: "negative warm", warm: -1, wantErr: true},
248+ {name: "negative idle", warm: 1, idle: -time.Second, wantErr: true},
249+ }
250+ for _, tt := range tests {
251+ t.Run(tt.name, func(t *testing.T) {
252+ pool, err := NewPool(1, portsmocks.NewMockBrokerRouteAuthority(t), portsmocks.NewMockBrokerIdentityBinder(t), portsmocks.NewMockBrokerEndpointConnector(t), newManualClock(time.Unix(0, 0)), PoolLimits{Physical: 1, Clients: 1, Streams: 1, StreamsPerClient: 1, Warm: tt.warm, Idle: tt.idle})
253+ if tt.wantErr {
254+ require.ErrorIs(t, err, ports.BrokerAdmissionInvalid)
255+ return
256+ }
257+ require.NoError(t, err)
258+ require.NoError(t, pool.Close())
259+ })
260+ }
261+}
M internal/usecase/broker/service_test.go +1 -1
@@ -39,7 +39,7 @@ func composeTestAuthority(t testing.TB, epoch ports.BrokerEpoch, registry *Regis
3939 resolver := poolResolver(func(_ context.Context, r ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
4040 return ports.BrokerDialTarget{Fence: ports.BrokerEndpointFence{Local: true}, Policy: r.Policy, Address: "fake", StartMode: r.StartMode, ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: "canonical", Bound: true}}, nil
4141 })
42- pool, err := NewPool(epoch, resolver, registry, connector, clock, PoolLimits{Physical: 4, Clients: 8, Streams: 16, StreamsPerClient: 16, Idle: time.Minute})
42+ pool, err := NewPool(epoch, resolver, registry, connector, clock, PoolLimits{Physical: 4, Clients: 8, Streams: 16, StreamsPerClient: 16, Warm: 4, Idle: time.Minute})
4343 require.NoError(t, err)
4444 t.Cleanup(func() { require.NoError(t, pool.Close()) })
4545 supervisor, idleClock := newTestSupervisor(t, time.Hour)
@@ -39,7 +39,7 @@ func composeTestAuthority(t testing.TB, epoch ports.BrokerEpoch, registry *Regis
39 resolver := poolResolver(func(_ context.Context, r ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {39 resolver := poolResolver(func(_ context.Context, r ports.BrokerOpenStreamRequest) (ports.BrokerDialTarget, error) {
40 return ports.BrokerDialTarget{Fence: ports.BrokerEndpointFence{Local: true}, Policy: r.Policy, Address: "fake", StartMode: r.StartMode, ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: "canonical", Bound: true}}, nil40 return ports.BrokerDialTarget{Fence: ports.BrokerEndpointFence{Local: true}, Policy: r.Policy, Address: "fake", StartMode: r.StartMode, ExpectedIdentity: ports.BrokerExpectedIdentity{Identity: "canonical", Bound: true}}, nil
41 })41 })
42- pool, err := NewPool(epoch, resolver, registry, connector, clock, PoolLimits{Physical: 4, Clients: 8, Streams: 16, StreamsPerClient: 16, Idle: time.Minute})42+ pool, err := NewPool(epoch, resolver, registry, connector, clock, PoolLimits{Physical: 4, Clients: 8, Streams: 16, StreamsPerClient: 16, Warm: 4, Idle: time.Minute})
43 require.NoError(t, err)43 require.NoError(t, err)
44 t.Cleanup(func() { require.NoError(t, pool.Close()) })44 t.Cleanup(func() { require.NoError(t, pool.Close()) })
45 supervisor, idleClock := newTestSupervisor(t, time.Hour)45 supervisor, idleClock := newTestSupervisor(t, time.Hour)
M scripts/remote-picker-harness/README.md +16 -10
@@ -41,22 +41,28 @@ committed contexts are captured: no host state, sockets, or key material.
4141 | `local-palette-cycle` | M1 local IPC: palette open/action/close, committed output, stable lifecycle |
4242 | `direct-remote-named@quic` | M2 direct named remote QUIC: exact fixture lifecycle, committed output, stable lifecycle |
4343 | `direct-remote-named@stdio` | M3 direct named remote explicit stdio: same semantics over SSH stdio |
44-| `warm-reuse@quic` | Warm hybrid QUIC: local→named remote→local→same remote in one Runner, same remote lifecycle, retained and new committed output, exactly one suspend/activate and one remote attach |
45-| `warm-reuse@stdio` | Warm hybrid stdio: same shape across the retained SSH process |
44+| `warm-reuse@quic` | Warm hybrid QUIC: local→named remote→local→same remote in one client, same remote lifecycle, retained and new committed output, no second broker dial |
45+| `warm-reuse@stdio` | Warm hybrid stdio: same shape across the broker's retained SSH process |
4646 | `hybrid-exact-return@quic` | M4 hybrid QUIC: local→remote→exact local return with committed action |
4747 | `hybrid-exact-return@stdio` | M5 hybrid stdio: same shape across close/dial |
4848 | `client-picker-navigate-local@quic` | M1: client-owned picker presented by the client, searched commit, cancel without a route change |
4949 | `client-picker-navigate-direct@quic` | M2: same over the remote serving daemon, no local daemon involved |
5050 | `client-picker-navigate-direct@stdio` | M3: same across stdio |
5151
52-Warm reuse asserts deterministic no-redial evidence rather than timing: the
53-client debug log must carry exactly one `remote attachment suspended` and one
54-`warm remote attachment activated` for the origin/session and zero
55-`warm remote activation failed`, and the remote daemon log must show exactly
56-one `client attached` for that session. A cold fallback or second bootstrap
57-(or, when the remote proxy log is available, a second proxy start) fails the
58-scenario. The remote committed lifecycle and the retained pane output before
59-and after the round trip are compared directly.
52+Warm reuse is owned by the connection broker, which replaced the client's
53+suspended-attachment cache (`remote.attachment-cache*`). Leaving a remote ends
54+its logical attachment; the broker keeps the physical SSH/QUIC transport warm
55+(`warmTransports` and `warmIdleTimeout` in `broker.json`, see
56+[configuration](../../docs/configuration.md#warm-remote-transports)), and the
57+return attaches afresh over it. The scenario asserts deterministic no-redial
58+evidence rather than timing: the client container's broker log
59+(`~/.local/state/vev/broker/log/vev-daemon.log`) must carry zero
60+`broker_remote_dial` events from the first remote commit to the end of the
61+journey, so a second bootstrap by the attach or by an observation probe fails
62+the scenario. The remote daemon's `client attached` count is recorded in the
63+evidence, not asserted: each visit is a new logical attachment. The remote
64+committed lifecycle and the retained pane output before and after the round
65+trip are compared directly.
6066
6167 A failed assertion fails the command. This is a targeted regression, not an
6268 exhaustive transport or geometry matrix. For a focused run, set
@@ -41,22 +41,28 @@ committed contexts are captured: no host state, sockets, or key material.
41 | `local-palette-cycle` | M1 local IPC: palette open/action/close, committed output, stable lifecycle |41 | `local-palette-cycle` | M1 local IPC: palette open/action/close, committed output, stable lifecycle |
42 | `direct-remote-named@quic` | M2 direct named remote QUIC: exact fixture lifecycle, committed output, stable lifecycle |42 | `direct-remote-named@quic` | M2 direct named remote QUIC: exact fixture lifecycle, committed output, stable lifecycle |
43 | `direct-remote-named@stdio` | M3 direct named remote explicit stdio: same semantics over SSH stdio |43 | `direct-remote-named@stdio` | M3 direct named remote explicit stdio: same semantics over SSH stdio |
44-| `warm-reuse@quic` | Warm hybrid QUIC: local→named remote→local→same remote in one Runner, same remote lifecycle, retained and new committed output, exactly one suspend/activate and one remote attach |44+| `warm-reuse@quic` | Warm hybrid QUIC: local→named remote→local→same remote in one client, same remote lifecycle, retained and new committed output, no second broker dial |
45-| `warm-reuse@stdio` | Warm hybrid stdio: same shape across the retained SSH process |45+| `warm-reuse@stdio` | Warm hybrid stdio: same shape across the broker's retained SSH process |
46 | `hybrid-exact-return@quic` | M4 hybrid QUIC: local→remote→exact local return with committed action |46 | `hybrid-exact-return@quic` | M4 hybrid QUIC: local→remote→exact local return with committed action |
47 | `hybrid-exact-return@stdio` | M5 hybrid stdio: same shape across close/dial |47 | `hybrid-exact-return@stdio` | M5 hybrid stdio: same shape across close/dial |
48 | `client-picker-navigate-local@quic` | M1: client-owned picker presented by the client, searched commit, cancel without a route change |48 | `client-picker-navigate-local@quic` | M1: client-owned picker presented by the client, searched commit, cancel without a route change |
49 | `client-picker-navigate-direct@quic` | M2: same over the remote serving daemon, no local daemon involved |49 | `client-picker-navigate-direct@quic` | M2: same over the remote serving daemon, no local daemon involved |
50 | `client-picker-navigate-direct@stdio` | M3: same across stdio |50 | `client-picker-navigate-direct@stdio` | M3: same across stdio |
51 51
52-Warm reuse asserts deterministic no-redial evidence rather than timing: the52+Warm reuse is owned by the connection broker, which replaced the client's
53-client debug log must carry exactly one `remote attachment suspended` and one53+suspended-attachment cache (`remote.attachment-cache*`). Leaving a remote ends
54-`warm remote attachment activated` for the origin/session and zero54+its logical attachment; the broker keeps the physical SSH/QUIC transport warm
55-`warm remote activation failed`, and the remote daemon log must show exactly55+(`warmTransports` and `warmIdleTimeout` in `broker.json`, see
56-one `client attached` for that session. A cold fallback or second bootstrap56+[configuration](../../docs/configuration.md#warm-remote-transports)), and the
57-(or, when the remote proxy log is available, a second proxy start) fails the57+return attaches afresh over it. The scenario asserts deterministic no-redial
58-scenario. The remote committed lifecycle and the retained pane output before58+evidence rather than timing: the client container's broker log
59-and after the round trip are compared directly.59+(`~/.local/state/vev/broker/log/vev-daemon.log`) must carry zero
60+`broker_remote_dial` events from the first remote commit to the end of the
61+journey, so a second bootstrap by the attach or by an observation probe fails
62+the scenario. The remote daemon's `client attached` count is recorded in the
63+evidence, not asserted: each visit is a new logical attachment. The remote
64+committed lifecycle and the retained pane output before and after the round
65+trip are compared directly.
60 66
61 A failed assertion fails the command. This is a targeted regression, not an67 A failed assertion fails the command. This is a targeted regression, not an
62 exhaustive transport or geometry matrix. For a focused run, set68 exhaustive transport or geometry matrix. For a focused run, set
M scripts/remote-picker-harness/acceptance.py +34 -48
@@ -14,10 +14,10 @@ reaches the driver process.
1414
1515 Direct-remote scenarios use ``--remote ENDPOINT --session NAME`` against a
1616 named fixture session: a named exact remote target, never an ephemeral
17-attach. The warm-reuse scenario keeps one client Runner across a
17+attach. The warm-reuse scenario keeps one client across a
1818 local -> named remote -> local -> same named remote journey and pins the
19-retained attachment with client debug logs, the remote daemon's attach
20-count, and the committed remote lifecycle.
19+broker's warm physical transport with the broker's remote-dial log, plus
20+the committed remote lifecycle and retained output.
2121 """
2222
2323 import json
@@ -36,6 +36,8 @@ STATE_LOG_DIR = "$HOME/.local/state/vev"
3636 CLIENT_LOG = "vev-client.log"
3737 DAEMON_LOG = "vev-daemon.log"
3838 STDIO_LOG = "vev-stdio.log"
39+# The broker logs one broker_remote_dial per remote physical bootstrap.
40+BROKER_LOG = "broker/log/vev-daemon.log"
3941 REMOTE_CACHE_FILE = "remote-catalog-cache.json"
4042 INVENTORY_TIMEOUT_S = 45
4143 ARTIFACTS_DIR = os.environ.get("VEV_ACCEPTANCE_ARTIFACTS_DIR", "").strip()
@@ -376,16 +378,18 @@ def scenario_direct_remote_named(client_container, remote_container, env):
376378
377379
378380 def scenario_warm_reuse(client_container, remote_container, env):
379- """Warm remote reuse across one client Runner.
380-
381- Journey: local -> named remote -> local -> same named remote. The remote
382- attachment is expected to suspend on the way back to local and reactivate
383- warm on the return: same committed remote lifecycle, retained pane output,
384- and new committed output. No-redial evidence is deterministic: the client
385- debug log records exactly one suspend and one warm activation (zero
386- fallbacks), and the remote daemon records exactly one attach for the
387- session, so a second bootstrap or dial would be visible. When the remote
388- proxy debug log is available it must also show exactly one proxy start.
381+ """Warm remote reuse through the broker's warm transport pool.
382+
383+ Journey: local -> named remote -> local -> same named remote. Leaving the
384+ remote ends its logical attachment, but the broker keeps the physical
385+ SSH/QUIC transport warm (broker.json warmTransports/warmIdleTimeout).
386+ The return attaches afresh over that transport: same committed remote
387+ lifecycle, retained pane output, and new committed output. No-redial
388+ evidence is deterministic: the broker log records zero broker_remote_dial
389+ events from the first remote commit to the end of the journey, so a second
390+ SSH bootstrap or QUIC handshake (by the attach or by an observation probe)
391+ fails the scenario. The remote daemon's attach count is recorded, not
392+ asserted: each visit is a new logical attachment by design.
389393 """
390394 mode = env.get("VEV_REMOTE_TRANSPORT", "quic")
391395 suffix = uuid.uuid4().hex[:8]
@@ -400,14 +404,13 @@ def scenario_warm_reuse(client_container, remote_container, env):
400404 remote_lifecycle = fixture_session["lifecycle_id"]
401405 # Snapshot log positions after the fixture so only events from this client
402406 # journey are counted.
403- client_offset = log_size(client_container, CLIENT_LOG)
404407 remote_offset = log_size(remote_container, DAEMON_LOG)
405- stdio_offset = log_size(remote_container, STDIO_LOG)
406408 driver = Driver(client_container, ["--session", local_name], env=env)
407409 evidence = {"scenario": "warm-reuse", "mode": mode,
408410 "local_session": local_name, "remote_session": remote_name,
409411 "remote_lifecycle": remote_lifecycle,
410412 "fixture_session": fixture_session}
413+ broker_offset = None
411414 try:
412415 initial = driver.call("capture")["context"]
413416 check(initial["session"]["session_name"] == local_name,
@@ -420,15 +423,17 @@ def scenario_warm_reuse(client_container, remote_container, env):
420423 check(first_remote["session"] == fixture_session,
421424 "first remote attach did not commit the exact named lifecycle", first_remote)
422425 evidence["first_remote"] = first_remote
426+ # Every dial after this point would be a re-bootstrap.
427+ broker_offset = log_size(client_container, BROKER_LOG)
423428 # Committed output while the remote pane is authoritative.
424429 commit_output(driver, "printf 'WARM_%s' PERSIST", "WARM_PERSIST")
425- # 2. remote -> local: the remote attachment suspends and stays warm.
430+ # 2. remote -> local: the logical attachment ends; the transport stays warm.
426431 switch_route(driver, local_name, "local", local_lifecycle)
427432 local_return = driver.call("capture")["context"]
428433 check(local_return["session"]["session_name"] == local_name,
429434 "return to local did not commit", local_return)
430435 evidence["local_return"] = local_return
431- # 3. local -> same named remote: the retained transport reactivates.
436+ # 3. local -> same named remote: a new attach over the warm transport.
432437 switch_route(driver, remote_name, "remote", remote_lifecycle)
433438 second_remote = driver.call("capture")["context"]
434439 check(second_remote["session"]["lifecycle_id"] == remote_lifecycle,
@@ -445,41 +450,22 @@ def scenario_warm_reuse(client_container, remote_container, env):
445450 evidence["after_warm_output"] = after_output
446451 finally:
447452 driver.close()
448- client_events = json_events(log_since(client_container, CLIENT_LOG, client_offset))
449- remote_events = json_events(log_since(remote_container, DAEMON_LOG, remote_offset))
450- stdio_events = json_events(log_since(remote_container, STDIO_LOG, stdio_offset))
451- evidence["log_counts"] = {
452- "client_suspended": count_events(
453- client_events, "remote attachment suspended",
454- origin="remote", session=remote_name),
455- "client_warm_activated": count_events(
456- client_events, "warm remote attachment activated",
457- origin="remote", session=remote_name, lifecycle=remote_lifecycle),
458- "client_warm_failed": count_events(
459- client_events, "warm remote activation failed",
460- origin="remote", session=remote_name),
461- "remote_client_attached": count_events(
462- remote_events, "client attached", session=remote_name),
463- "remote_proxy_started": (
464- count_events(stdio_events, "stdio proxy starting") +
465- count_events(stdio_events, "quic proxy starting")),
466- }
453+ if broker_offset is not None:
454+ broker_events = json_events(log_since(client_container, BROKER_LOG, broker_offset))
455+ remote_events = json_events(log_since(remote_container, DAEMON_LOG, remote_offset))
456+ evidence["log_counts"] = {
457+ "broker_remote_dials_after_first_attach": count_events(
458+ broker_events, "broker_remote_dial"),
459+ "remote_client_attached": count_events(
460+ remote_events, "client attached", session=remote_name),
461+ }
467462 write_artifact(f"warm-reuse-{mode}.json", evidence)
468463 counts = evidence["log_counts"]
469- check(counts["client_suspended"] == 1,
470- "expected exactly one remote attachment suspension", counts)
471- check(counts["client_warm_activated"] == 1,
472- "expected exactly one warm remote activation", counts)
473- check(counts["client_warm_failed"] == 0,
474- "warm activation failed or fell back to a cold dial", counts)
475- check(counts["remote_client_attached"] == 1,
476- "remote daemon saw a second bootstrap/attach (redial)", counts)
477- if counts["remote_proxy_started"]:
478- check(counts["remote_proxy_started"] == 1,
479- "remote proxy started more than once (redial)", counts)
464+ check(counts["broker_remote_dials_after_first_attach"] == 0,
465+ "broker re-bootstrapped the remote instead of reusing the warm transport", counts)
480466 print(f"PASS warm-reuse mode={mode} local={local_name} remote={remote_name} "
481467 f"lifecycle={remote_lifecycle[:8]} "
482- f"attaches={counts['remote_client_attached']}")
468+ f"attaches={counts['remote_client_attached']} redials=0")
483469
484470
485471 def picker_topology(env):
@@ -14,10 +14,10 @@ reaches the driver process.
14 14
15 Direct-remote scenarios use ``--remote ENDPOINT --session NAME`` against a15 Direct-remote scenarios use ``--remote ENDPOINT --session NAME`` against a
16 named fixture session: a named exact remote target, never an ephemeral16 named fixture session: a named exact remote target, never an ephemeral
17-attach. The warm-reuse scenario keeps one client Runner across a17+attach. The warm-reuse scenario keeps one client across a
18 local -> named remote -> local -> same named remote journey and pins the18 local -> named remote -> local -> same named remote journey and pins the
19-retained attachment with client debug logs, the remote daemon's attach19+broker's warm physical transport with the broker's remote-dial log, plus
20-count, and the committed remote lifecycle.20+the committed remote lifecycle and retained output.
21 """21 """
22 22
23 import json23 import json
@@ -36,6 +36,8 @@ STATE_LOG_DIR = "$HOME/.local/state/vev"
36 CLIENT_LOG = "vev-client.log"36 CLIENT_LOG = "vev-client.log"
37 DAEMON_LOG = "vev-daemon.log"37 DAEMON_LOG = "vev-daemon.log"
38 STDIO_LOG = "vev-stdio.log"38 STDIO_LOG = "vev-stdio.log"
39+# The broker logs one broker_remote_dial per remote physical bootstrap.
40+BROKER_LOG = "broker/log/vev-daemon.log"
39 REMOTE_CACHE_FILE = "remote-catalog-cache.json"41 REMOTE_CACHE_FILE = "remote-catalog-cache.json"
40 INVENTORY_TIMEOUT_S = 4542 INVENTORY_TIMEOUT_S = 45
41 ARTIFACTS_DIR = os.environ.get("VEV_ACCEPTANCE_ARTIFACTS_DIR", "").strip()43 ARTIFACTS_DIR = os.environ.get("VEV_ACCEPTANCE_ARTIFACTS_DIR", "").strip()
@@ -376,16 +378,18 @@ def scenario_direct_remote_named(client_container, remote_container, env):
376 378
377 379
378 def scenario_warm_reuse(client_container, remote_container, env):380 def scenario_warm_reuse(client_container, remote_container, env):
379- """Warm remote reuse across one client Runner.381+ """Warm remote reuse through the broker's warm transport pool.
380-382+
381- Journey: local -> named remote -> local -> same named remote. The remote383+ Journey: local -> named remote -> local -> same named remote. Leaving the
382- attachment is expected to suspend on the way back to local and reactivate384+ remote ends its logical attachment, but the broker keeps the physical
383- warm on the return: same committed remote lifecycle, retained pane output,385+ SSH/QUIC transport warm (broker.json warmTransports/warmIdleTimeout).
384- and new committed output. No-redial evidence is deterministic: the client386+ The return attaches afresh over that transport: same committed remote
385- debug log records exactly one suspend and one warm activation (zero387+ lifecycle, retained pane output, and new committed output. No-redial
386- fallbacks), and the remote daemon records exactly one attach for the388+ evidence is deterministic: the broker log records zero broker_remote_dial
387- session, so a second bootstrap or dial would be visible. When the remote389+ events from the first remote commit to the end of the journey, so a second
388- proxy debug log is available it must also show exactly one proxy start.390+ SSH bootstrap or QUIC handshake (by the attach or by an observation probe)
391+ fails the scenario. The remote daemon's attach count is recorded, not
392+ asserted: each visit is a new logical attachment by design.
389 """393 """
390 mode = env.get("VEV_REMOTE_TRANSPORT", "quic")394 mode = env.get("VEV_REMOTE_TRANSPORT", "quic")
391 suffix = uuid.uuid4().hex[:8]395 suffix = uuid.uuid4().hex[:8]
@@ -400,14 +404,13 @@ def scenario_warm_reuse(client_container, remote_container, env):
400 remote_lifecycle = fixture_session["lifecycle_id"]404 remote_lifecycle = fixture_session["lifecycle_id"]
401 # Snapshot log positions after the fixture so only events from this client405 # Snapshot log positions after the fixture so only events from this client
402 # journey are counted.406 # journey are counted.
403- client_offset = log_size(client_container, CLIENT_LOG)
404 remote_offset = log_size(remote_container, DAEMON_LOG)407 remote_offset = log_size(remote_container, DAEMON_LOG)
405- stdio_offset = log_size(remote_container, STDIO_LOG)
406 driver = Driver(client_container, ["--session", local_name], env=env)408 driver = Driver(client_container, ["--session", local_name], env=env)
407 evidence = {"scenario": "warm-reuse", "mode": mode,409 evidence = {"scenario": "warm-reuse", "mode": mode,
408 "local_session": local_name, "remote_session": remote_name,410 "local_session": local_name, "remote_session": remote_name,
409 "remote_lifecycle": remote_lifecycle,411 "remote_lifecycle": remote_lifecycle,
410 "fixture_session": fixture_session}412 "fixture_session": fixture_session}
413+ broker_offset = None
411 try:414 try:
412 initial = driver.call("capture")["context"]415 initial = driver.call("capture")["context"]
413 check(initial["session"]["session_name"] == local_name,416 check(initial["session"]["session_name"] == local_name,
@@ -420,15 +423,17 @@ def scenario_warm_reuse(client_container, remote_container, env):
420 check(first_remote["session"] == fixture_session,423 check(first_remote["session"] == fixture_session,
421 "first remote attach did not commit the exact named lifecycle", first_remote)424 "first remote attach did not commit the exact named lifecycle", first_remote)
422 evidence["first_remote"] = first_remote425 evidence["first_remote"] = first_remote
426+ # Every dial after this point would be a re-bootstrap.
427+ broker_offset = log_size(client_container, BROKER_LOG)
423 # Committed output while the remote pane is authoritative.428 # Committed output while the remote pane is authoritative.
424 commit_output(driver, "printf 'WARM_%s' PERSIST", "WARM_PERSIST")429 commit_output(driver, "printf 'WARM_%s' PERSIST", "WARM_PERSIST")
425- # 2. remote -> local: the remote attachment suspends and stays warm.430+ # 2. remote -> local: the logical attachment ends; the transport stays warm.
426 switch_route(driver, local_name, "local", local_lifecycle)431 switch_route(driver, local_name, "local", local_lifecycle)
427 local_return = driver.call("capture")["context"]432 local_return = driver.call("capture")["context"]
428 check(local_return["session"]["session_name"] == local_name,433 check(local_return["session"]["session_name"] == local_name,
429 "return to local did not commit", local_return)434 "return to local did not commit", local_return)
430 evidence["local_return"] = local_return435 evidence["local_return"] = local_return
431- # 3. local -> same named remote: the retained transport reactivates.436+ # 3. local -> same named remote: a new attach over the warm transport.
432 switch_route(driver, remote_name, "remote", remote_lifecycle)437 switch_route(driver, remote_name, "remote", remote_lifecycle)
433 second_remote = driver.call("capture")["context"]438 second_remote = driver.call("capture")["context"]
434 check(second_remote["session"]["lifecycle_id"] == remote_lifecycle,439 check(second_remote["session"]["lifecycle_id"] == remote_lifecycle,
@@ -445,41 +450,22 @@ def scenario_warm_reuse(client_container, remote_container, env):
445 evidence["after_warm_output"] = after_output450 evidence["after_warm_output"] = after_output
446 finally:451 finally:
447 driver.close()452 driver.close()
448- client_events = json_events(log_since(client_container, CLIENT_LOG, client_offset))453+ if broker_offset is not None:
449- remote_events = json_events(log_since(remote_container, DAEMON_LOG, remote_offset))454+ broker_events = json_events(log_since(client_container, BROKER_LOG, broker_offset))
450- stdio_events = json_events(log_since(remote_container, STDIO_LOG, stdio_offset))455+ remote_events = json_events(log_since(remote_container, DAEMON_LOG, remote_offset))
451- evidence["log_counts"] = {456+ evidence["log_counts"] = {
452- "client_suspended": count_events(457+ "broker_remote_dials_after_first_attach": count_events(
453- client_events, "remote attachment suspended",458+ broker_events, "broker_remote_dial"),
454- origin="remote", session=remote_name),459+ "remote_client_attached": count_events(
455- "client_warm_activated": count_events(460+ remote_events, "client attached", session=remote_name),
456- client_events, "warm remote attachment activated",461+ }
457- origin="remote", session=remote_name, lifecycle=remote_lifecycle),
458- "client_warm_failed": count_events(
459- client_events, "warm remote activation failed",
460- origin="remote", session=remote_name),
461- "remote_client_attached": count_events(
462- remote_events, "client attached", session=remote_name),
463- "remote_proxy_started": (
464- count_events(stdio_events, "stdio proxy starting") +
465- count_events(stdio_events, "quic proxy starting")),
466- }
467 write_artifact(f"warm-reuse-{mode}.json", evidence)462 write_artifact(f"warm-reuse-{mode}.json", evidence)
468 counts = evidence["log_counts"]463 counts = evidence["log_counts"]
469- check(counts["client_suspended"] == 1,464+ check(counts["broker_remote_dials_after_first_attach"] == 0,
470- "expected exactly one remote attachment suspension", counts)465+ "broker re-bootstrapped the remote instead of reusing the warm transport", counts)
471- check(counts["client_warm_activated"] == 1,
472- "expected exactly one warm remote activation", counts)
473- check(counts["client_warm_failed"] == 0,
474- "warm activation failed or fell back to a cold dial", counts)
475- check(counts["remote_client_attached"] == 1,
476- "remote daemon saw a second bootstrap/attach (redial)", counts)
477- if counts["remote_proxy_started"]:
478- check(counts["remote_proxy_started"] == 1,
479- "remote proxy started more than once (redial)", counts)
480 print(f"PASS warm-reuse mode={mode} local={local_name} remote={remote_name} "466 print(f"PASS warm-reuse mode={mode} local={local_name} remote={remote_name} "
481 f"lifecycle={remote_lifecycle[:8]} "467 f"lifecycle={remote_lifecycle[:8]} "
482- f"attaches={counts['remote_client_attached']}")468+ f"attaches={counts['remote_client_attached']} redials=0")
483 469
484 470
485 def picker_topology(env):471 def picker_topology(env):

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 12

Tags 11

↑↓ moveenter openesc close