1// Integration coverage for the real SSH stdio daemonmux carriage.
2//
3// These tests exercise the daemonmux multiplexer over an actual SSH stdio
4// carriage: a real owned subprocess whose stdio is the physical carriage,
5// dialed by sshstdio.DialMuxContext and served by the helper half,
6// sshstdio.NewStdioTransport, not over the in-memory carrier used by the unit
7// tests or the socket/QUIC carriages. One subprocess is one physical daemonmux
8// connection: every logical stream rides its stdio, and no logical attachment
9// is ever mapped to a second process.
10//
11// The broker half (EndpointConnector, PhysicalConnection, LogicalConnector)
12// runs in the test process exactly as it would in the client. The daemon half
13// runs in a re-executed copy of this test binary (the helper process), so the
14// process start, pipe ownership, reaping, and teardown are real; no production
15// composition is started. For the raw pump-accounting checks the daemon-side
16// pump runs in the test process over the subprocess stdio relay so the stream
17// accounting can be asserted directly, exactly as the Unix and QUIC carriage
18// tests do.
19package daemonmux
20
21import (
22 "context"
23 "errors"
24 "flag"
25 "fmt"
26 "io"
27 "net"
28 "os"
29 "path/filepath"
30 "runtime"
31 "strconv"
32 "strings"
33 "sync"
34 "sync/atomic"
35 "testing"
36 "time"
37
38 "github.com/stretchr/testify/require"
39
40 "github.com/bnema/vev/internal/adapters/sshstdio"
41 "github.com/bnema/vev/internal/domain"
42 "github.com/bnema/vev/internal/ports"
43 "github.com/bnema/vev/internal/protocol"
44 "github.com/bnema/vev/internal/protocol/wire"
45)
46
47// sshMuxCarriageDeadline bounds every real subprocess carriage wait.
48const sshMuxCarriageDeadline = 10 * time.Second
49
50// sshMuxHelperMarker distinguishes the re-executed helper process from an
51// ordinary test run. It travels as a positional argument, so a normal test run
52// (no positional args) skips every helper entry point.
53const sshMuxHelperMarker = "vev-daemonmux-ssh-carriage-helper"
54
55// TestSSHMuxHelperProcess is the daemon (or relay) half of one SSH stdio
56// carriage. It is inert unless re-executed by a carriage test with the helper
57// marker, at which point its stdio is the physical carriage and it must never
58// write anything but framed envelopes to stdout.
59func TestSSHMuxHelperProcess(t *testing.T) {
60 args := flag.Args()
61 if len(args) < 2 || args[0] != sshMuxHelperMarker {
62 t.Skip("ssh stdio mux helper process")
63 }
64 switch mode := args[1]; mode {
65 case "daemon":
66 runSSHMuxDaemonHelper(t, sshMuxHelperArg(args, 2))
67 case "relay":
68 runSSHMuxRelayHelper(sshMuxHelperArg(args, 2))
69 case "silent":
70 runSSHMuxSilentHelper()
71 default:
72 fmt.Fprintf(os.Stderr, "daemonmux ssh mux helper: unknown mode %q\n", mode)
73 os.Exit(2)
74 }
75}
76
77func sshMuxHelperArg(args []string, index int) string {
78 if index < len(args) {
79 return args[index]
80 }
81 return ""
82}
83
84// runSSHMuxDaemonHelper serves one daemonmux physical connection over this
85// process' own stdin/stdout using the helper-side stdio carriage constructor
86// and the real ServerSupervisor/AggregateListener/typed Listener path. It stops
87// when the carriage ends and never lets the test framework's report reach the
88// carriage's stdout, which would corrupt the framing stream.
89func runSSHMuxDaemonHelper(t *testing.T, pidPath string) {
90 if pidPath != "" {
91 if err := os.WriteFile(pidPath, []byte(strconv.Itoa(os.Getpid())), 0o600); err != nil {
92 sshMuxHelperFatal(err)
93 }
94 }
95 raw := sshstdio.NewStdioTransport()
96 binding := mustServerBinding(t)
97 aggregate := NewAggregateListener()
98 supervisor, err := NewServerSupervisor(aggregate, binding, DefaultMuxCeilings(), 0)
99 sshMuxHelperFatal(err)
100 sshMuxHelperFatal(supervisor.Adopt(context.Background(), raw))
101 go func() {
102 for {
103 conn, err := aggregate.Accept()
104 if err != nil {
105 return
106 }
107 go func() { _ = servePongs(conn) }()
108 }
109 }()
110 if pump := sshMuxHelperPump(supervisor); pump != nil {
111 <-pump.Done()
112 }
113 _ = supervisor.Close()
114 _ = aggregate.Close()
115 sshMuxSilenceStdout()
116}
117
118// sshMuxHelperPump returns the supervisor's one owned child pump, so the helper
119// exits exactly when the physical carriage reaches its terminal outcome.
120func sshMuxHelperPump(s *ServerSupervisor) *Pump {
121 s.mu.Lock()
122 defer s.mu.Unlock()
123 for child := range s.children {
124 return child.pump
125 }
126 return nil
127}
128
129// runSSHMuxRelayHelper relays this process' stdio and one parent-bound Unix
130// socket transparently. It lets the raw pump tests keep both mux ends in the
131// test process while every byte still crosses a real subprocess pipe.
132func runSSHMuxRelayHelper(sockPath string) {
133 if sockPath == "" {
134 sshMuxHelperFatal(errors.New("relay: missing socket path"))
135 }
136 conn, err := net.Dial("unix", sockPath)
137 sshMuxHelperFatal(err)
138 go func() {
139 _, _ = io.Copy(conn, os.Stdin)
140 _ = conn.Close()
141 }()
142 _, _ = io.Copy(os.Stdout, conn)
143 _ = conn.Close()
144 sshMuxSilenceStdout()
145}
146
147// runSSHMuxSilentHelper reads its stdin to EOF without ever answering, so a
148// setup deadline is the only way a caller can give up. It exits cleanly once the
149// caller closes the carriage.
150func runSSHMuxSilentHelper() {
151 _, _ = io.Copy(io.Discard, os.Stdin)
152 sshMuxSilenceStdout()
153}
154
155// sshMuxHelperFatal reports a helper failure on stderr (the bounded diagnostic
156// channel) and exits non-cleanly.
157func sshMuxHelperFatal(err error) {
158 if err == nil {
159 return
160 }
161 fmt.Fprintln(os.Stderr, "daemonmux ssh mux helper:", err)
162 os.Exit(2)
163}
164
165// sshMuxSilenceStdout points the test framework's final report at /dev/null so
166// a helper that returns normally never writes its report into the carriage.
167func sshMuxSilenceStdout() {
168 devnull, err := os.OpenFile(os.DevNull, os.O_WRONLY, 0)
169 if err == nil {
170 os.Stdout = devnull
171 }
172}
173
174// sshMuxHelperCommand builds an explicit command specification that re-executes
175// this test binary in one helper mode. The command is never the session
176// `_broker-mux-stdio` command: the mux carriage takes its whole command from the caller.
177func sshMuxHelperCommand(mode string, extra ...string) sshstdio.CommandSpec {
178 args := append([]string{"-test.run=^TestSSHMuxHelperProcess$", sshMuxHelperMarker, mode}, extra...)
179 return sshstdio.CommandSpec{Path: os.Args[0], Args: args}
180}
181
182// sshMuxDialer returns a RawCarrierDialer over one explicit command
183// specification and a counter of how many physical carriages it opened.
184func sshMuxDialer(spec sshstdio.CommandSpec) (RawCarrierDialer, *atomic.Int32) {
185 dials := &atomic.Int32{}
186 return func(ctx context.Context, _ ports.BrokerDialTarget) (RawFramedTransport, error) {
187 dials.Add(1)
188 return sshstdio.DialMuxContext(ctx, spec, nil)
189 }, dials
190}
191
192// sshMuxConnect dials one helper carriage and returns the live broker physical
193// connection, registering its teardown.
194func sshMuxConnect(t *testing.T, binding ServerBinding, spec sshstdio.CommandSpec) ports.BrokerPhysicalConnection {
195 t.Helper()
196 dial, _ := sshMuxDialer(spec)
197 connector, err := NewEndpointConnector(dial, DefaultMuxCeilings())
198 require.NoError(t, err)
199 ctx, cancel := context.WithTimeout(context.Background(), sshMuxCarriageDeadline)
200 defer cancel()
201 physical, err := connector.Connect(ctx, muxEndpoint(binding.Identity(), binding.Policy(), "ssh://helper"))
202 require.NoError(t, err)
203 t.Cleanup(func() { _ = physical.Close() })
204 return physical
205}
206
207// sshMuxReadPID waits for the helper to publish its process id.
208func sshMuxReadPID(t *testing.T, path string) int {
209 t.Helper()
210 pid := 0
211 require.Eventually(t, func() bool {
212 data, err := os.ReadFile(path)
213 if err != nil {
214 return false
215 }
216 value, err := strconv.Atoi(strings.TrimSpace(string(data)))
217 if err != nil {
218 return false
219 }
220 pid = value
221 return true
222 }, sshMuxCarriageDeadline, 2*time.Millisecond)
223 return pid
224}
225
226// requireNoGoroutineLeak waits for the test process to return to its baseline
227// goroutine count once every carriage was torn down.
228func requireNoGoroutineLeak(t *testing.T, baseline int) {
229 t.Helper()
230 require.Eventually(t, func() bool { return runtime.NumGoroutine() <= baseline+4 }, 5*time.Second, 20*time.Millisecond,
231 "goroutines leaked: baseline %d, now %d", baseline, runtime.NumGoroutine())
232}
233
234// TestSSHCarriageSupervisorTwoThenHundredTypedAdmissions drives the full daemon
235// path over one real SSH stdio physical connection: EndpointConnector dials a
236// real subprocess started by DialMuxContext, the helper process adopts it with
237// ServerSupervisor/AggregateListener, and the typed Listener delivers
238// independent daemon admissions. Only one physical connection and one process
239// exist for all 102 streams.
240func TestSSHCarriageSupervisorTwoThenHundredTypedAdmissions(t *testing.T) {
241 baseline := runtime.NumGoroutine()
242 binding := mustServerBinding(t)
243 policy := binding.Policy()
244 pidPath := filepath.Join(t.TempDir(), "helper.pid")
245 spec := sshMuxHelperCommand("daemon", pidPath)
246
247 dial, dials := sshMuxDialer(spec)
248 connector, err := NewEndpointConnector(dial, DefaultMuxCeilings())
249 require.NoError(t, err)
250 physical, err := connector.Connect(context.Background(), muxEndpoint(binding.Identity(), policy, "ssh://helper"))
251 require.NoError(t, err)
252 t.Cleanup(func() { _ = physical.Close() })
253 pid := sshMuxReadPID(t, pidPath)
254
255 // Two independent typed admissions over the one physical carriage.
256 first, err := openTyped(physical, context.Background(), muxOpenRequest(1, policy))
257 require.NoError(t, err)
258 t.Cleanup(func() { _ = first.Close() })
259 second, err := openTyped(physical, context.Background(), muxOpenRequest(2, policy))
260 require.NoError(t, err)
261 t.Cleanup(func() { _ = second.Close() })
262 for _, stream := range []ports.BrokerLogicalConnection{first, second} {
263 require.NoError(t, stream.SendClient(protocol.Ping{}))
264 message, err := stream.ReceiveServer()
265 require.NoError(t, err)
266 require.Equal(t, protocol.Pong{}, message)
267 }
268
269 // One hundred more independently admitted typed streams, opened
270 // concurrently over the same physical carriage.
271 const extra = 100
272 connections := make([]ports.BrokerLogicalConnection, extra)
273 openErrs := make([]error, extra)
274 var wg sync.WaitGroup
275 for i := 0; i < extra; i++ {
276 wg.Add(1)
277 go func(i int) {
278 defer wg.Done()
279 connections[i], openErrs[i] = openTyped(physical, context.Background(), muxOpenRequest(uint64(3+i), policy))
280 }(i)
281 }
282 wg.Wait()
283 for i, openErr := range openErrs {
284 require.NoError(t, openErr, "stream %d", i+3)
285 }
286 for i, connection := range connections {
287 require.NotNil(t, connection, "stream %d", i+3)
288 t.Cleanup(func() { _ = connection.Close() })
289 }
290 for i, connection := range connections {
291 require.NoError(t, connection.SendClient(protocol.Ping{}), "stream %d", i+3)
292 message, err := connection.ReceiveServer()
293 require.NoError(t, err, "stream %d", i+3)
294 require.Equal(t, protocol.Pong{}, message)
295 }
296
297 // All 102 streams share exactly one physical connection: exactly one dial,
298 // one subprocess, and one live broker carriage.
299 require.Equal(t, int32(1), dials.Load(), "all streams must share one ssh carriage")
300 require.False(t, channelClosed(physical.Done()), "the physical carriage must stay open while streams are live")
301
302 all := make([]ports.BrokerLogicalConnection, 0, 2+extra)
303 all = append(all, first, second)
304 all = append(all, connections...)
305 for _, connection := range all {
306 require.NoError(t, connection.Close())
307 }
308 require.False(t, channelClosed(physical.Done()), "closing every stream must leave the physical carriage open")
309
310 // Teardown terminates and reaps the one owned subprocess, and no worker
311 // outlives the carriage.
312 require.NoError(t, physical.Close())
313 require.Eventually(t, func() bool { return sshMuxProcessGone(pid) }, sshMuxCarriageDeadline, 5*time.Millisecond,
314 "the ssh carriage subprocess was never reaped")
315 requireNoGoroutineLeak(t, baseline)
316}
317
318// TestSSHCarriagePhysicalFailureFansOutAndIsIsolated proves a lost SSH stdio
319// physical carriage publishes its terminal outcome before fanning out to every
320// logical stream it carried, while a sibling physical carriage on the same
321// supervisor is untouched and a fresh stream on the lost physical is refused.
322func TestSSHCarriagePhysicalFailureFansOutAndIsIsolated(t *testing.T) {
323 binding := mustServerBinding(t)
324 policy := binding.Policy()
325 pidA := filepath.Join(t.TempDir(), "a.pid")
326 pidB := filepath.Join(t.TempDir(), "b.pid")
327
328 physicalA := sshMuxConnect(t, binding, sshMuxHelperCommand("daemon", pidA))
329 physicalB := sshMuxConnect(t, binding, sshMuxHelperCommand("daemon", pidB))
330 lostPID := sshMuxReadPID(t, pidA)
331 _ = sshMuxReadPID(t, pidB)
332
333 lost := make([]ports.BrokerLogicalConnection, 3)
334 for i := range lost {
335 connection, err := openTyped(physicalA, context.Background(), muxOpenRequest(uint64(i+1), policy))
336 require.NoError(t, err, "stream %d", i+1)
337 lost[i] = connection
338 t.Cleanup(func() { _ = connection.Close() })
339 }
340 survivor, err := openTyped(physicalB, context.Background(), muxOpenRequest(1, policy))
341 require.NoError(t, err)
342 t.Cleanup(func() { _ = survivor.Close() })
343
344 // The physical terminal outcome is ordered before its logical fan-out.
345 ordered := make(chan bool, 1)
346 go func() {
347 select {
348 case <-lost[0].Done():
349 case <-time.After(sshMuxCarriageDeadline):
350 ordered <- false
351 return
352 }
353 select {
354 case <-physicalA.Done():
355 ordered <- true
356 case <-time.After(sshMuxCarriageDeadline):
357 ordered <- false
358 }
359 }()
360
361 // Lose the ssh carriage entirely: the subprocess dies, its stdout closes,
362 // and the broker observes the pipe failure.
363 process, err := os.FindProcess(lostPID)
364 require.NoError(t, err)
365 require.NoError(t, process.Kill())
366
367 require.True(t, <-ordered, "the logical Done fired before the physical Done")
368 require.True(t, channelClosed(physicalA.Done()))
369 require.Error(t, physicalA.Err())
370 require.NotEqual(t, domain.RemoteFailureNone, physicalA.FailureKind())
371 for i, stream := range lost {
372 require.Eventually(t, func() bool {
373 return channelClosed(stream.Done()) && stream.Err() != nil
374 }, sshMuxCarriageDeadline, time.Millisecond, "lost stream %d never received the physical failure", i+1)
375 }
376
377 // The sibling physical carriage is unaffected.
378 require.False(t, channelClosed(physicalB.Done()))
379 require.False(t, channelClosed(survivor.Done()))
380 require.NoError(t, survivor.SendClient(protocol.Ping{}))
381 message, err := survivor.ReceiveServer()
382 require.NoError(t, err)
383 require.Equal(t, protocol.Pong{}, message)
384
385 // The lost physical refuses a fresh stream.
386 _, err = openTyped(physicalA, context.Background(), muxOpenRequest(9, policy))
387 require.Error(t, err)
388
389 // The failed subprocess is reaped, not left a zombie.
390 require.Eventually(t, func() bool { return sshMuxProcessGone(lostPID) }, sshMuxCarriageDeadline, 5*time.Millisecond,
391 "the failed ssh carriage subprocess was never reaped")
392}
393
394// TestSSHCarriageSetupContextDetachedFromPhysicalLifetime proves the setup
395// context covers only setup: once EndpointConnector.Connect returned over a real
396// ssh carriage, cancelling the setup context never stops the pooled physical
397// connection, which Close alone owns.
398func TestSSHCarriageSetupContextDetachedFromPhysicalLifetime(t *testing.T) {
399 binding := mustServerBinding(t)
400 spec := sshMuxHelperCommand("daemon", filepath.Join(t.TempDir(), "helper.pid"))
401 dial, _ := sshMuxDialer(spec)
402 connector, err := NewEndpointConnector(dial, DefaultMuxCeilings())
403 require.NoError(t, err)
404
405 setupCtx, cancel := context.WithCancel(context.Background())
406 physical, err := connector.Connect(setupCtx, muxEndpoint(binding.Identity(), binding.Policy(), "ssh://helper"))
407 require.NoError(t, err)
408 t.Cleanup(func() { _ = physical.Close() })
409 cancel()
410
411 require.False(t, channelClosed(physical.Done()))
412 logical, err := openTyped(physical, context.Background(), muxOpenRequest(1, binding.Policy()))
413 require.NoError(t, err)
414 t.Cleanup(func() { _ = logical.Close() })
415 require.NoError(t, logical.SendClient(protocol.Ping{}))
416 message, err := logical.ReceiveServer()
417 require.NoError(t, err)
418 require.Equal(t, protocol.Pong{}, message)
419 require.False(t, channelClosed(physical.Done()))
420}
421
422// TestSSHCarriageCanceledSetupAdmitsNothing proves one canceled setup context
423// prevents the subprocess from starting at all: no process, no physical
424// connection, no stream.
425func TestSSHCarriageCanceledSetupAdmitsNothing(t *testing.T) {
426 binding := mustServerBinding(t)
427 pidPath := filepath.Join(t.TempDir(), "helper.pid")
428 spec := sshMuxHelperCommand("daemon", pidPath)
429 dial, _ := sshMuxDialer(spec)
430 connector, err := NewEndpointConnector(dial, DefaultMuxCeilings())
431 require.NoError(t, err)
432
433 setupCtx, cancel := context.WithCancel(context.Background())
434 cancel()
435 physical, err := connector.Connect(setupCtx, muxEndpoint(binding.Identity(), binding.Policy(), "ssh://helper"))
436 require.ErrorIs(t, err, context.Canceled)
437 require.Nil(t, physical)
438
439 _, statErr := os.Stat(pidPath)
440 require.ErrorIs(t, statErr, os.ErrNotExist, "a canceled setup must not start the ssh subprocess")
441}
442
443// TestSSHCarriageSetupDeadlineBoundsTheMuxHandshake proves the setup deadline
444// covers the daemonmux physical handshake over a real ssh carriage: a helper
445// that never answers the preamble fails the whole Connect within the setup
446// deadline and publishes no physical connection.
447func TestSSHCarriageSetupDeadlineBoundsTheMuxHandshake(t *testing.T) {
448 binding := mustServerBinding(t)
449 spec := sshMuxHelperCommand("silent")
450 dial, _ := sshMuxDialer(spec)
451 connector, err := NewEndpointConnector(dial, DefaultMuxCeilings())
452 require.NoError(t, err)
453
454 setupCtx, cancel := context.WithTimeout(context.Background(), 300*time.Millisecond)
455 defer cancel()
456 physical, err := connector.Connect(setupCtx, muxEndpoint(binding.Identity(), binding.Policy(), "ssh://helper"))
457 require.ErrorIs(t, err, context.DeadlineExceeded)
458 require.Nil(t, physical)
459}
460
461// TestSSHCarriageConnectErrorBoundedAndSanitized proves a failed ssh carriage
462// surfaces a bounded, control-byte-free public error that never carries raw
463// remote stderr.
464func TestSSHCarriageConnectErrorBoundedAndSanitized(t *testing.T) {
465 const marker = "remote-secret-value"
466 binding := mustServerBinding(t)
467 script := "sleep 0.2; printf '\\033[31m" + marker + "\\033[0m' >&2; printf 'B%.0s' $(seq 1 20000) >&2; exit 3"
468 spec := sshstdio.CommandSpec{Path: "sh", Args: []string{"-c", script}}
469 dial, _ := sshMuxDialer(spec)
470 connector, err := NewEndpointConnector(dial, DefaultMuxCeilings())
471 require.NoError(t, err)
472
473 ctx, cancel := context.WithTimeout(context.Background(), sshMuxCarriageDeadline)
474 defer cancel()
475 physical, err := connector.Connect(ctx, muxEndpoint(binding.Identity(), binding.Policy(), "ssh://helper"))
476 require.Error(t, err)
477 require.Nil(t, physical)
478
479 cause := errors.Unwrap(err)
480 if cause == nil {
481 cause = err
482 }
483 require.NotContains(t, cause.Error(), marker)
484 require.NotContains(t, cause.Error(), "\x1b")
485 require.Less(t, len(cause.Error()), 4096)
486}
487
488// sshRawCarriagePair establishes one real SSH stdio carriage whose subprocess is
489// a transparent relay to an in-process Unix socket, and returns its two raw
490// ends: the ssh client end (the subprocess stdio DialMuxContext owns) and the
491// in-process relay end. Every byte crosses the real subprocess pipes.
492func sshRawCarriagePair(t *testing.T) (RawFramedTransport, RawFramedTransport) {
493 t.Helper()
494 sockPath := filepath.Join(filepath.Dir(muxCarriagePath(t)), "relay.sock")
495 listener, err := net.Listen("unix", sockPath)
496 require.NoError(t, err)
497 t.Cleanup(func() { _ = listener.Close() })
498
499 accepted := make(chan net.Conn, 1)
500 acceptErr := make(chan error, 1)
501 go func() {
502 conn, err := listener.Accept()
503 if err != nil {
504 acceptErr <- err
505 return
506 }
507 accepted <- conn
508 }()
509
510 client, err := sshstdio.DialMuxContext(context.Background(), sshMuxHelperCommand("relay", sockPath), nil)
511 require.NoError(t, err)
512 t.Cleanup(func() { _ = client.Close() })
513
514 var conn net.Conn
515 select {
516 case conn = <-accepted:
517 case err := <-acceptErr:
518 t.Fatalf("relay accept: %v", err)
519 case <-time.After(sshMuxCarriageDeadline):
520 t.Fatal("relay subprocess never connected")
521 }
522 t.Cleanup(func() { _ = conn.Close() })
523 daemon := sshstdio.NewTransport(conn, conn, conn.Close).(wire.BoundedTransport)
524 return client, daemon
525}
526
527// sshDaemonPump builds a daemon-side pump over one end of a real SSH stdio
528// carriage and returns the raw ssh client end for direct frame injection, so a
529// test can drive deterministic inbound traffic over the actual subprocess
530// pipes.
531func sshDaemonPump(t *testing.T) (*Pump, RawFramedTransport) {
532 t.Helper()
533 client, daemon := sshRawCarriagePair(t)
534 carrier, err := NewPreambleCarrier(daemon)
535 require.NoError(t, err)
536 require.NoError(t, carrier.Negotiate(DefaultMuxCeilings()))
537 pump, err := NewPump(carrier, DirectionClient, DefaultMuxCeilings())
538 require.NoError(t, err)
539 t.Cleanup(func() { _ = pump.Close() })
540 pump.Start(context.Background())
541 return pump, client
542}
543
544// injectSSHClientFrame writes one client frame over the real ssh carriage.
545func injectSSHClientFrame(t *testing.T, client RawFramedTransport, message ClientMessage) {
546 t.Helper()
547 require.NoError(t, client.Send(wire.Envelope{Payload: mustEncodeClient(t, message)}))
548}
549
550// nextSSHServerFrame reads one daemon frame from the real ssh carriage.
551func nextSSHServerFrame(t *testing.T, client RawFramedTransport) []byte {
552 t.Helper()
553 envelope, err := client.RecvBounded(testEnvelopeCeiling)
554 require.NoError(t, err)
555 return envelope.Payload
556}
557
558// TestSSHCarriageBlockedConsumerSiblingProgress proves, over a real ssh
559// carriage, that a logical consumer which never drains one stream's inbound
560// queue neither blocks the reader nor a sibling: the stalled stream is reset by
561// its own bound while the sibling keeps receiving.
562func TestSSHCarriageBlockedConsumerSiblingProgress(t *testing.T) {
563 pump, client := sshDaemonPump(t)
564
565 const siblings = 2
566 for i := 1; i <= siblings; i++ {
567 injectSSHClientFrame(t, client, openFor(PhysicalStreamID(i)))
568 }
569 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == siblings })
570 for i := 1; i <= siblings; i++ {
571 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(PhysicalStreamID(i))}))
572 }
573
574 // A peer ignores flow control and fills stream 1's whole window while the
575 // consumer never Takes it.
576 chunk, full := floodWindow(pump.Ceilings())
577 for i := 0; i < full; i++ {
578 injectSSHClientFrame(t, client, Data{Physical: 1, Data: chunk})
579 }
580 requireEngineEventually(t, pump, func(e *StreamEngine) bool {
581 status, ok := e.Status(1)
582 return ok && status.QueuedChunks == full
583 })
584
585 // One more chunk exceeds the granted window; a sibling chunk still lands,
586 // so neither the reader nor the sibling was blocked.
587 injectSSHClientFrame(t, client, Data{Physical: 1, Data: chunk})
588 injectSSHClientFrame(t, client, Data{Physical: 2, Data: []byte("sibling")})
589
590 taken, ok := takeEventually(t, pump, 2)
591 require.True(t, ok)
592 require.Equal(t, []byte("sibling"), taken)
593
594 stalled := mustStatus(t, pump.Engine(), 1)
595 require.Equal(t, StreamTerminal, stalled.State)
596 require.ErrorIs(t, stalled.Err, ErrStreamQueueFull)
597 require.Equal(t, domain.RemoteFailureInvalidResponse, stalled.FailureKind)
598 require.Equal(t, StreamOpen, mustStatus(t, pump.Engine(), 2).State)
599 require.False(t, pump.Engine().Closed())
600 require.False(t, channelClosed(pump.Done()))
601
602 reset := decodeReset(t, nextSSHServerFrame(t, client), DirectionServer)
603 require.Equal(t, PhysicalStreamID(1), reset.Physical)
604 require.Equal(t, domain.RemoteFailureInvalidResponse, reset.Error.FailureKind)
605}
606
607// TestSSHCarriageResetIsolation proves, over a real ssh carriage, that one
608// stream-local refusal schedules exactly one Reset for its own stream and
609// leaves the physical connection and every sibling healthy.
610func TestSSHCarriageResetIsolation(t *testing.T) {
611 pump, client := sshDaemonPump(t)
612
613 for _, id := range []PhysicalStreamID{1, 2, 3} {
614 injectSSHClientFrame(t, client, openFor(id))
615 }
616 requireEngineEventually(t, pump, func(e *StreamEngine) bool { return e.Live() == 3 })
617 for _, id := range []PhysicalStreamID{2, 3} {
618 require.NoError(t, pump.Engine().Opened(Opened{Ref: testRef(id)}))
619 }
620
621 // Stream 1 is still opening: inbound data is a stream-local state refusal.
622 injectSSHClientFrame(t, client, Data{Physical: 1, Data: []byte("early")})
623 requireEngineEventually(t, pump, func(e *StreamEngine) bool {
624 status, ok := e.Status(1)
625 return ok && status.State == StreamTerminal
626 })
627 settled := mustStatus(t, pump.Engine(), 1)
628 require.ErrorContains(t, settled.Err, "invalid stream state")
629 require.Equal(t, domain.RemoteFailureInvalidResponse, settled.FailureKind)
630
631 injectSSHClientFrame(t, client, Data{Physical: 2, Data: []byte("sibling")})
632 taken, ok := takeEventually(t, pump, 2)
633 require.True(t, ok)
634 require.Equal(t, []byte("sibling"), taken)
635
636 reset := decodeReset(t, nextSSHServerFrame(t, client), DirectionServer)
637 require.Equal(t, PhysicalStreamID(1), reset.Physical)
638 require.True(t, reset.HasError)
639 require.Equal(t, domain.RemoteFailureInvalidResponse, reset.Error.FailureKind)
640 require.False(t, pump.Engine().Closed())
641 require.False(t, channelClosed(pump.Done()))
642}