From 80929aa2d5588c11194ed48f5827c3fd250b5b35 Mon Sep 17 00:00:00 2001 From: Omry Yadan Date: Sun, 9 Aug 2026 04:24:29 +0800 Subject: [PATCH] Bridge controlled-session controller and workload I/O Add a backend-neutral bridge that dispatches typed controller requests through lifecycle authorization, applies accepted PTY input and resize operations, and forwards exact ordered output with bounded backpressure. Give validated lifecycle events a separate prioritized write admission path without allowing callers to forge opened or output events. Preserve explicit cancellation, disconnect, and output-finalization diagnostics while leaving lifecycle and resource ownership to the host supervisor. Cover authorization rejection, binary output, lifecycle priority and cancellation, slow consumers, disconnects, resize, ordinary Ctrl-C, and exact Docker exit behavior with race-enabled unit and live integration tests. Update the controlled-session implementation status. --- docs/CONTROLLED_SESSION_DESIGN.md | 28 +- internal/controlledsession/channel.go | 10 +- .../controlledsession/channel_linux_test.go | 23 +- internal/controlledsession/session_io.go | 305 ++++++++++ internal/controlledsession/session_io_test.go | 549 ++++++++++++++++++ ...led_session_session_io_integration_test.go | 317 ++++++++++ 6 files changed, 1219 insertions(+), 13 deletions(-) create mode 100644 internal/controlledsession/session_io.go create mode 100644 internal/controlledsession/session_io_test.go create mode 100644 internal/dockerdeploy/controlled_session_session_io_integration_test.go diff --git a/docs/CONTROLLED_SESSION_DESIGN.md b/docs/CONTROLLED_SESSION_DESIGN.md index d34cccc..dad61a3 100644 --- a/docs/CONTROLLED_SESSION_DESIGN.md +++ b/docs/CONTROLLED_SESSION_DESIGN.md @@ -38,9 +38,16 @@ summary: Capability-scoped execution sessions that inherit Reploy's global conta before container creation, creates the frozen controller plan inert, starts it at most once, independently observes its exit, exposes graceful and forced stop operations, captures the full container ID returned by creation, and - pins all later lifecycle operations to that exact container. Protocol/PTY - bridging, controlled-session networking, and lifecycle orchestration remain - later slices. + pins all later lifecycle operations to that exact container. The + backend-neutral session I/O bridge is implemented: it dispatches typed + controller requests to an injected lifecycle handler, applies only + lifecycle-accepted input and resize effects, forwards exact ordered PTY + output as protocol events, gives lifecycle events a separate prioritized + bounded write admission path, and reports request, backpressure, and + disconnect failures without owning the channel or containers. A failed event + write makes the framed transport terminal so a later event cannot be appended + to a potentially partial frame. Full lifecycle orchestration and + controlled-session networking remain later slices. - Initial runtime: Linux containers under Docker - Motivating clients: OmegaFlow recording, sandboxed AI agents, security inspection, and untrusted-code execution @@ -543,11 +550,16 @@ If every final byte is delivered and every output surface reaches EOF before the deadline, Host Reploy emits `workload_outputs_finalized(drained)` only after all earlier output has been consumed through its flow-control window. If the deadline expires, or a runtime error makes complete delivery unverifiable, -Host Reploy forcibly closes the remaining surfaces and emits -`workload_outputs_finalized(failed, reason)`. The multiplexing layer guarantees -that no output frame can follow either outcome. Failure is explicit and cannot -be converted into successful completion by `complete` or terminal -acknowledgement. +Host Reploy forcibly closes the remaining surfaces, records the failed outcome, +and emits `workload_outputs_finalized(failed, reason)` while the controller +transport remains intact. Any event-frame write failure is terminal: the frame +header or payload may have been written partially, so Host Reploy closes the +connection and never appends another event. That failure latches +`controller_lost`; the output-finalization failure remains in the invoking host +operation's result even though the damaged channel cannot carry it. The +multiplexing layer guarantees that no output frame can follow a successfully +emitted finalization outcome. Failure cannot be converted into successful +completion by `complete` or terminal acknowledgement. The barrier initially covers the PTY. A future workload output-file or output-directory contract joins the same barrier after its files are closed, diff --git a/internal/controlledsession/channel.go b/internal/controlledsession/channel.go index b45cfee..32f7010 100644 --- a/internal/controlledsession/channel.go +++ b/internal/controlledsession/channel.go @@ -76,6 +76,8 @@ type PrivateChannelV1 struct { // ControllerConnectionV1 is the sole claimed controller connection. Reads and // writes each admit one frame at a time, so no unbounded in-memory queue forms. +// A failed event write closes the connection because it may have left a partial +// frame on the stream. type ControllerConnectionV1 struct { connection net.Conn readMu sync.Mutex @@ -234,10 +236,14 @@ func (connection *ControllerConnectionV1) WriteEvent(ctx context.Context, event return WriteEventV1(connection.connection, event) }) if err != nil { + closeErr := connection.Close() if isControllerDisconnectV1(err) { - return fmt.Errorf("%w while writing event: %v", ErrControllerDisconnectedV1, err) + return errors.Join( + fmt.Errorf("%w while writing event: %v", ErrControllerDisconnectedV1, err), + closeErr, + ) } - return fmt.Errorf("write controlled-session event: %w", err) + return errors.Join(fmt.Errorf("write controlled-session event: %w", err), closeErr) } return nil } diff --git a/internal/controlledsession/channel_linux_test.go b/internal/controlledsession/channel_linux_test.go index 75148d5..c7f34a8 100644 --- a/internal/controlledsession/channel_linux_test.go +++ b/internal/controlledsession/channel_linux_test.go @@ -7,6 +7,7 @@ import ( "context" "encoding/binary" "errors" + "io" "net" "os" "path/filepath" @@ -250,11 +251,27 @@ func TestControllerConnectionV1BoundsAndSerializesFlow(t *testing.T) { server, client := net.Pipe() defer client.Close() connection := &ControllerConnectionV1{connection: server} - ctx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond) + ctx, cancel := context.WithTimeout(context.Background(), 250*time.Millisecond) defer cancel() event := EventV1{Kind: EventOutputV1, Bytes: bytes.Repeat([]byte{'x'}, MaxFramePayloadV1)} - if err := connection.WriteEvent(ctx, event); !errors.Is(err, context.DeadlineExceeded) { - t.Fatalf("WriteEvent() error = %v", err) + writeErr := make(chan error, 1) + go func() { writeErr <- connection.WriteEvent(ctx, event) }() + header := make([]byte, frameHeaderSizeV1) + if _, err := io.ReadFull(client, header); err != nil { + t.Fatal(err) + } + if err := <-writeErr; !errors.Is(err, context.DeadlineExceeded) { + t.Fatalf("first WriteEvent() error = %v", err) + } + if err := connection.WriteEvent(t.Context(), EventV1{Kind: EventOutputV1, Bytes: []byte("later")}); err == nil { + t.Fatal("second WriteEvent() reused the connection after a partial frame") + } + remaining, err := io.ReadAll(client) + if err != nil { + t.Fatal(err) + } + if len(remaining) != 0 { + t.Fatalf("partial payload bytes = %d, want 0", len(remaining)) } }) diff --git a/internal/controlledsession/session_io.go b/internal/controlledsession/session_io.go new file mode 100644 index 0000000..28544cc --- /dev/null +++ b/internal/controlledsession/session_io.go @@ -0,0 +1,305 @@ +package controlledsession + +import ( + "context" + "errors" + "fmt" + "io" + "sync" + "time" +) + +// ControllerTransportV1 is the claimed, typed controller channel used by one +// controlled session. Implementations must admit at most one request read and +// one event write at a time and make both operations cancelable. Any event +// write failure is terminal because the framed stream may contain a partial +// frame; implementations must not admit another event write afterward. +type ControllerTransportV1 interface { + ReadRequest(context.Context) (RequestV1, error) + WriteEvent(context.Context, EventV1) error +} + +type sessionEventWriteGateV1 struct { + transport ControllerTransportV1 + + mu sync.Mutex + writing bool + lifecycleWaiting int + writeFailure error + changed chan struct{} +} + +// WorkloadPTYControlV1 is the backend-neutral part of a workload PTY that can +// receive already-authorized controller operations. +type WorkloadPTYControlV1 interface { + WriteInput(context.Context, []byte) error + Resize(context.Context, uint32, uint32) error +} + +// ControllerRequestHandlerV1 applies lifecycle authorization and the effects +// of one typed controller request. It must return only after the request has +// been accepted or rejected and must stop promptly when ctx is canceled. +type ControllerRequestHandlerV1 func(context.Context, RequestV1) error + +// ApplyAcceptedWorkloadPTYRequestV1 applies only the PTY effects of a request +// that the lifecycle supervisor has already accepted. It returns false for +// lifecycle-only requests so their effects remain owned by that supervisor. +func ApplyAcceptedWorkloadPTYRequestV1( + ctx context.Context, + workload WorkloadPTYControlV1, + request RequestV1, +) (bool, error) { + if ctx == nil || ctx.Done() == nil { + return false, fmt.Errorf("apply accepted controlled-session workload PTY request: cancelable context is required") + } + if workload == nil { + return false, fmt.Errorf("apply accepted controlled-session workload PTY request: workload is required") + } + if err := ValidateRequestV1(request); err != nil { + return false, fmt.Errorf("apply accepted controlled-session workload PTY request: %w", err) + } + switch request.Kind { + case RequestInputV1: + if err := workload.WriteInput(ctx, request.Bytes); err != nil { + return true, fmt.Errorf("apply accepted controlled-session input request: %w", err) + } + return true, nil + case RequestResizeV1: + if err := workload.Resize(ctx, request.Columns, request.Rows); err != nil { + return true, fmt.Errorf("apply accepted controlled-session resize request: %w", err) + } + return true, nil + case RequestTerminateV1, RequestCompleteV1, RequestAcknowledgeTerminatedV1: + return false, nil + default: + panic("validated controlled-session request has an unsupported kind") + } +} + +// SessionIOBridgeV1 connects one claimed controller channel to one workload +// PTY without owning lifecycle policy, the channel, or the workload container. +// Requests remain serialized through the injected handler. PTY output is +// delivered as exact ordered protocol events with the output pump's single +// bounded flow-control chunk. +type SessionIOBridgeV1 struct { + transport ControllerTransportV1 + eventWrite *sessionEventWriteGateV1 + output *PTYOutputPumpV1 + + requestCtx context.Context + cancelRequest context.CancelFunc + requestDone chan struct{} + stopOnce sync.Once + + requestResultMu sync.Mutex + requestResult error +} + +// StartSessionIOBridgeV1 starts request dispatch and PTY output delivery +// immediately. Callers may therefore establish the bridge before starting the +// workload, preventing early output loss. +func StartSessionIOBridgeV1( + transport ControllerTransportV1, + output io.ReadCloser, + handle ControllerRequestHandlerV1, +) (*SessionIOBridgeV1, error) { + if transport == nil { + return nil, fmt.Errorf("start controlled-session I/O bridge: controller transport is required") + } + if handle == nil { + return nil, fmt.Errorf("start controlled-session I/O bridge: controller request handler is required") + } + requestCtx, cancelRequest := context.WithCancel(context.Background()) + bridge := &SessionIOBridgeV1{ + transport: transport, + eventWrite: newSessionEventWriteGateV1(transport), + requestCtx: requestCtx, + cancelRequest: cancelRequest, + requestDone: make(chan struct{}), + } + pump, err := StartPTYOutputPumpV1(output, func(ctx context.Context, bytes []byte) error { + return bridge.eventWrite.write(ctx, EventV1{Kind: EventOutputV1, Bytes: bytes}, false) + }) + if err != nil { + cancelRequest() + return nil, fmt.Errorf("start controlled-session I/O bridge output: %w", err) + } + bridge.output = pump + go bridge.runRequests(handle) + return bridge, nil +} + +// SendLifecycleEvent gives one lifecycle event priority over the next output +// frame while preserving the frame already being written. Opened remains owned +// by channel claim, output remains owned by the PTY pump, and final event-order +// decisions remain the lifecycle supervisor's responsibility. +func (bridge *SessionIOBridgeV1) SendLifecycleEvent(ctx context.Context, event EventV1) error { + if !isBridgeLifecycleEventV1(event.Kind) { + return fmt.Errorf("send controlled-session lifecycle event: event kind %q is not lifecycle-owned", event.Kind) + } + if err := ValidateEventV1(event); err != nil { + return fmt.Errorf("send controlled-session lifecycle event: %w", err) + } + if err := bridge.eventWrite.write(ctx, event, true); err != nil { + return fmt.Errorf("send controlled-session lifecycle event: %w", err) + } + return nil +} + +// RequestsDone closes after request dispatch stops. The caller must inspect +// WaitRequests; closure alone does not mean that the controller completed. +func (bridge *SessionIOBridgeV1) RequestsDone() <-chan struct{} { + return bridge.requestDone +} + +// StopRequests cancels request dispatch without closing the controller channel. +// It is idempotent and is treated as normal host-owned shutdown by WaitRequests. +func (bridge *SessionIOBridgeV1) StopRequests() { + bridge.stopOnce.Do(bridge.cancelRequest) +} + +// WaitRequests waits for request dispatch to stop and returns its immutable +// diagnostic. Caller cancellation stops only this wait. +func (bridge *SessionIOBridgeV1) WaitRequests(ctx context.Context) error { + if ctx == nil || ctx.Done() == nil { + return fmt.Errorf("wait for controlled-session controller requests: cancelable context is required") + } + select { + case <-bridge.requestDone: + bridge.requestResultMu.Lock() + defer bridge.requestResultMu.Unlock() + return bridge.requestResult + case <-ctx.Done(): + return ctx.Err() + } +} + +func (bridge *SessionIOBridgeV1) OutputDone() <-chan struct{} { + return bridge.output.Done() +} + +func (bridge *SessionIOBridgeV1) FinalizeOutput(deadline time.Time) (PTYOutputFinalizationV1, error) { + return bridge.output.Finalize(deadline) +} + +func (bridge *SessionIOBridgeV1) runRequests(handle ControllerRequestHandlerV1) { + defer close(bridge.requestDone) + for { + request, err := bridge.transport.ReadRequest(bridge.requestCtx) + if err != nil { + bridge.setRequestResult(bridge.requestFailure("read", err)) + return + } + if err := handle(bridge.requestCtx, request); err != nil { + bridge.setRequestResult(bridge.requestFailure("handle", err)) + return + } + } +} + +func (bridge *SessionIOBridgeV1) requestFailure(action string, err error) error { + if errors.Is(err, context.Canceled) && bridge.requestCtx.Err() != nil { + return nil + } + return fmt.Errorf("%s controlled-session controller request: %w", action, err) +} + +func (bridge *SessionIOBridgeV1) setRequestResult(err error) { + bridge.requestResultMu.Lock() + defer bridge.requestResultMu.Unlock() + bridge.requestResult = err +} + +func newSessionEventWriteGateV1(transport ControllerTransportV1) *sessionEventWriteGateV1 { + return &sessionEventWriteGateV1{transport: transport, changed: make(chan struct{})} +} + +func (gate *sessionEventWriteGateV1) write(ctx context.Context, event EventV1, lifecycle bool) error { + if err := gate.acquire(ctx, lifecycle); err != nil { + return err + } + err := gate.transport.WriteEvent(ctx, event) + gate.release(err) + return err +} + +func (gate *sessionEventWriteGateV1) acquire(ctx context.Context, lifecycle bool) error { + if ctx == nil || ctx.Done() == nil { + return fmt.Errorf("cancelable event context is required") + } + if err := ctx.Err(); err != nil { + return err + } + gate.mu.Lock() + if lifecycle { + gate.lifecycleWaiting++ + gate.notifyLocked() + } + for gate.writeFailure == nil && (gate.writing || !lifecycle && gate.lifecycleWaiting > 0) { + changed := gate.changed + gate.mu.Unlock() + select { + case <-changed: + case <-ctx.Done(): + gate.mu.Lock() + if lifecycle { + gate.lifecycleWaiting-- + gate.notifyLocked() + } + gate.mu.Unlock() + return ctx.Err() + } + gate.mu.Lock() + } + if gate.writeFailure != nil { + if lifecycle { + gate.lifecycleWaiting-- + gate.notifyLocked() + } + err := gate.writeFailure + gate.mu.Unlock() + return err + } + if err := ctx.Err(); err != nil { + if lifecycle { + gate.lifecycleWaiting-- + gate.notifyLocked() + } + gate.mu.Unlock() + return err + } + if lifecycle { + gate.lifecycleWaiting-- + } + gate.writing = true + gate.notifyLocked() + gate.mu.Unlock() + return nil +} + +func (gate *sessionEventWriteGateV1) release(writeErr error) { + gate.mu.Lock() + if writeErr != nil && gate.writeFailure == nil { + gate.writeFailure = fmt.Errorf("controlled-session event transport is unusable after write failure: %w", writeErr) + } + gate.writing = false + gate.notifyLocked() + gate.mu.Unlock() +} + +func (gate *sessionEventWriteGateV1) notifyLocked() { + close(gate.changed) + gate.changed = make(chan struct{}) +} + +func isBridgeLifecycleEventV1(kind EventKindV1) bool { + switch kind { + case EventWorkloadExitV1, EventTerminatingV1, EventDiagnosticV1, + EventWorkloadOutputsFinalizedV1, EventTerminatedV1: + return true + case EventOpenedV1, EventOutputV1: + return false + default: + return false + } +} diff --git a/internal/controlledsession/session_io_test.go b/internal/controlledsession/session_io_test.go new file mode 100644 index 0000000..e1ebf71 --- /dev/null +++ b/internal/controlledsession/session_io_test.go @@ -0,0 +1,549 @@ +package controlledsession + +import ( + "bytes" + "context" + "errors" + "io" + "reflect" + "strings" + "sync" + "testing" + "time" +) + +func TestApplyAcceptedWorkloadPTYRequestV1AppliesOnlyPTYEffects(t *testing.T) { + workload := &recordingWorkloadPTYControlV1{} + input := []byte{0x00, 0x03, 0x7f, 0xff} + tests := []struct { + request RequestV1 + handled bool + }{ + {request: RequestV1{Kind: RequestInputV1, Bytes: input}, handled: true}, + {request: RequestV1{Kind: RequestResizeV1, Columns: 132, Rows: 43}, handled: true}, + {request: RequestV1{Kind: RequestTerminateV1}}, + {request: RequestV1{Kind: RequestCompleteV1}}, + {request: RequestV1{Kind: RequestAcknowledgeTerminatedV1}}, + } + for _, test := range tests { + handled, err := ApplyAcceptedWorkloadPTYRequestV1(t.Context(), workload, test.request) + if err != nil { + t.Fatalf("ApplyAcceptedWorkloadPTYRequestV1(%s): %v", test.request.Kind, err) + } + if handled != test.handled { + t.Fatalf("ApplyAcceptedWorkloadPTYRequestV1(%s) handled = %t", test.request.Kind, handled) + } + } + if !bytes.Equal(workload.input, input) { + t.Fatalf("workload input = %v", workload.input) + } + if workload.columns != 132 || workload.rows != 43 { + t.Fatalf("workload dimensions = %dx%d", workload.columns, workload.rows) + } +} + +func TestApplyAcceptedWorkloadPTYRequestV1ReportsValidationAndBackendFailures(t *testing.T) { + if _, err := ApplyAcceptedWorkloadPTYRequestV1(context.Background(), &recordingWorkloadPTYControlV1{}, RequestV1{Kind: RequestTerminateV1}); err == nil { + t.Fatal("ApplyAcceptedWorkloadPTYRequestV1() accepted a non-cancelable context") + } + if _, err := ApplyAcceptedWorkloadPTYRequestV1(t.Context(), nil, RequestV1{Kind: RequestTerminateV1}); err == nil { + t.Fatal("ApplyAcceptedWorkloadPTYRequestV1() accepted a missing workload") + } + workload := &recordingWorkloadPTYControlV1{inputErr: errors.New("input failed"), resizeErr: errors.New("resize failed")} + if _, err := ApplyAcceptedWorkloadPTYRequestV1(t.Context(), workload, RequestV1{Kind: RequestInputV1}); err == nil { + t.Fatal("ApplyAcceptedWorkloadPTYRequestV1() accepted invalid input") + } + if _, err := ApplyAcceptedWorkloadPTYRequestV1(t.Context(), workload, RequestV1{Kind: RequestInputV1, Bytes: []byte("x")}); !errors.Is(err, workload.inputErr) { + t.Fatalf("input error = %v", err) + } + if _, err := ApplyAcceptedWorkloadPTYRequestV1(t.Context(), workload, RequestV1{Kind: RequestResizeV1, Columns: 80, Rows: 24}); !errors.Is(err, workload.resizeErr) { + t.Fatalf("resize error = %v", err) + } +} + +func TestSessionIOBridgeV1DispatchesRequestsAndOrderedBinaryOutput(t *testing.T) { + transport := newBridgeTestTransportV1() + reader, writer := io.Pipe() + workload := &recordingWorkloadPTYControlV1{} + handled := make(chan RequestV1, 5) + bridge, err := StartSessionIOBridgeV1(transport, reader, func(ctx context.Context, request RequestV1) error { + ptyRequest, err := ApplyAcceptedWorkloadPTYRequestV1(ctx, workload, request) + if err != nil { + return err + } + if ptyRequest && request.Kind != RequestInputV1 && request.Kind != RequestResizeV1 { + return errors.New("lifecycle request was reported as a PTY request") + } + handled <- request + return nil + }) + if err != nil { + t.Fatal(err) + } + + requests := []RequestV1{ + {Kind: RequestInputV1, Bytes: []byte{0x00, 0x03, 0xff}}, + {Kind: RequestResizeV1, Columns: 120, Rows: 40}, + {Kind: RequestTerminateV1}, + {Kind: RequestCompleteV1}, + {Kind: RequestAcknowledgeTerminatedV1}, + } + for _, request := range requests { + transport.requests <- bridgeTestRequestResultV1{request: request} + } + for index, want := range requests { + select { + case got := <-handled: + if !reflect.DeepEqual(got, want) { + t.Fatalf("handled request %d = %#v, want %#v", index, got, want) + } + case <-time.After(time.Second): + t.Fatalf("timed out waiting for request %d", index) + } + } + + payload := []byte{0x00, 0x01, 0x7f, 0xff, 'R', 'P', 'S', 'N'} + writeErr := make(chan error, 1) + go func() { + _, err := writer.Write(payload) + writeErr <- errors.Join(err, writer.Close()) + }() + select { + case event := <-transport.events: + if event.Kind != EventOutputV1 || !bytes.Equal(event.Bytes, payload) { + t.Fatalf("output event = %#v", event) + } + case <-time.After(time.Second): + t.Fatal("timed out waiting for output event") + } + if err := <-writeErr; err != nil { + t.Fatal(err) + } + result, err := bridge.FinalizeOutput(time.Now().Add(time.Second)) + if err != nil || result.Status.Kind != WorkloadOutputFinalizationDrainedV1 || result.Err != nil { + t.Fatalf("FinalizeOutput() = %#v, %v", result, err) + } + bridge.StopRequests() + waitCtx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := bridge.WaitRequests(waitCtx); err != nil { + t.Fatalf("WaitRequests() = %v", err) + } + if !bytes.Equal(workload.input, requests[0].Bytes) || workload.columns != 120 || workload.rows != 40 { + t.Fatalf("workload effects = input %v, dimensions %dx%d", workload.input, workload.columns, workload.rows) + } +} + +func TestSessionIOBridgeV1SurfacesDisconnectAndHandlerFailure(t *testing.T) { + t.Run("disconnect", func(t *testing.T) { + transport := newBridgeTestTransportV1() + bridge, err := StartSessionIOBridgeV1(transport, io.NopCloser(bytes.NewReader(nil)), func(context.Context, RequestV1) error { + return nil + }) + if err != nil { + t.Fatal(err) + } + disconnect := errors.New("controller disconnected") + transport.requests <- bridgeTestRequestResultV1{err: disconnect} + waitCtx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := bridge.WaitRequests(waitCtx); !errors.Is(err, disconnect) { + t.Fatalf("WaitRequests() = %v", err) + } + }) + + t.Run("handler", func(t *testing.T) { + transport := newBridgeTestTransportV1() + handleErr := errors.New("request rejected") + bridge, err := StartSessionIOBridgeV1(transport, io.NopCloser(bytes.NewReader(nil)), func(context.Context, RequestV1) error { + return handleErr + }) + if err != nil { + t.Fatal(err) + } + transport.requests <- bridgeTestRequestResultV1{request: RequestV1{Kind: RequestTerminateV1}} + waitCtx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := bridge.WaitRequests(waitCtx); !errors.Is(err, handleErr) { + t.Fatalf("WaitRequests() = %v", err) + } + }) +} + +func TestSessionIOBridgeV1LifecycleRejectionPreventsPTYEffect(t *testing.T) { + authorization := testAuthorizationV1() + authorization.Operations = []OperationV1{OperationResizeV1, OperationTerminateV1} + machine, err := NewMachineV1(authorization) + if err != nil { + t.Fatal(err) + } + if _, err := machine.Observe(ObservationV1{Kind: ObservationActivatedV1}); err != nil { + t.Fatal(err) + } + transport := newBridgeTestTransportV1() + workload := &recordingWorkloadPTYControlV1{} + bridge, err := StartSessionIOBridgeV1(transport, io.NopCloser(bytes.NewReader(nil)), func(ctx context.Context, request RequestV1) error { + if _, err := machine.ApplyRequest(request); err != nil { + return err + } + _, err := ApplyAcceptedWorkloadPTYRequestV1(ctx, workload, request) + return err + }) + if err != nil { + t.Fatal(err) + } + transport.requests <- bridgeTestRequestResultV1{request: RequestV1{Kind: RequestInputV1, Bytes: []byte("denied")}} + waitCtx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := bridge.WaitRequests(waitCtx); !errors.Is(err, ErrRequestRejected) { + t.Fatalf("WaitRequests() = %v", err) + } + if len(workload.input) != 0 { + t.Fatalf("rejected input reached the workload: %q", workload.input) + } +} + +func TestSessionIOBridgeV1BackpressureIsBoundedAndCancelable(t *testing.T) { + transport := newBridgeTestTransportV1() + transport.writeEntered = make(chan struct{}, 1) + transport.blockWrites = true + reader, writer := io.Pipe() + bridge, err := StartSessionIOBridgeV1(transport, reader, func(context.Context, RequestV1) error { return nil }) + if err != nil { + t.Fatal(err) + } + writeErr := make(chan error, 1) + go func() { + _, err := writer.Write(bytes.Repeat([]byte("x"), ptyOutputChunkSizeV1*2)) + writeErr <- err + }() + select { + case <-transport.writeEntered: + case <-time.After(time.Second): + t.Fatal("output delivery did not reach the bounded write") + } + select { + case <-bridge.OutputDone(): + t.Fatal("output pump crossed blocked backpressure") + default: + } + result, err := bridge.FinalizeOutput(time.Now().Add(20 * time.Millisecond)) + if err != nil || result.Status.Kind != WorkloadOutputFinalizationFailedV1 || !errors.Is(result.Err, context.DeadlineExceeded) { + t.Fatalf("FinalizeOutput() = %#v, %v", result, err) + } + if err := <-writeErr; err != nil && !errors.Is(err, io.ErrClosedPipe) { + t.Fatalf("blocked PTY writer error = %v", err) + } + _ = writer.Close() + bridge.StopRequests() +} + +func TestSessionIOBridgeV1PrioritizesLifecycleEventsBetweenOutputFrames(t *testing.T) { + transport := newPriorityBridgeTestTransportV1() + reader, writer := io.Pipe() + bridge, err := StartSessionIOBridgeV1(transport, reader, func(context.Context, RequestV1) error { return nil }) + if err != nil { + t.Fatal(err) + } + payload := bytes.Repeat([]byte("x"), ptyOutputChunkSizeV1*2) + writeErr := make(chan error, 1) + go func() { + _, err := writer.Write(payload) + writeErr <- errors.Join(err, writer.Close()) + }() + select { + case <-transport.firstWriteEntered: + case <-time.After(time.Second): + t.Fatal("first output event was not admitted") + } + lifecycle := EventV1{Kind: EventDiagnosticV1, Diagnostic: &DiagnosticV1{Code: "test", Message: "test diagnostic"}} + lifecycleErr := make(chan error, 1) + go func() { lifecycleErr <- bridge.SendLifecycleEvent(t.Context(), lifecycle) }() + waitForLifecycleEventAdmissionV1(t, bridge) + close(transport.releaseFirstWrite) + if err := <-lifecycleErr; err != nil { + t.Fatal(err) + } + if err := <-writeErr; err != nil { + t.Fatal(err) + } + result, err := bridge.FinalizeOutput(time.Now().Add(time.Second)) + if err != nil || result.Status.Kind != WorkloadOutputFinalizationDrainedV1 { + t.Fatalf("FinalizeOutput() = %#v, %v", result, err) + } + transport.mu.Lock() + events := append([]EventV1(nil), transport.events...) + transport.mu.Unlock() + if len(events) != 3 || events[0].Kind != EventOutputV1 || events[1].Kind != EventDiagnosticV1 || events[2].Kind != EventOutputV1 { + t.Fatalf("event order = %#v", events) + } + bridge.StopRequests() +} + +func TestSessionIOBridgeV1RemovesCanceledLifecycleEventAdmission(t *testing.T) { + transport := newPriorityBridgeTestTransportV1() + reader, writer := io.Pipe() + bridge, err := StartSessionIOBridgeV1(transport, reader, func(context.Context, RequestV1) error { return nil }) + if err != nil { + t.Fatal(err) + } + writeErr := make(chan error, 1) + go func() { + _, err := writer.Write([]byte("output")) + writeErr <- errors.Join(err, writer.Close()) + }() + select { + case <-transport.firstWriteEntered: + case <-time.After(time.Second): + t.Fatal("output event was not admitted") + } + lifecycleCtx, cancelLifecycle := context.WithCancel(context.Background()) + lifecycleErr := make(chan error, 1) + go func() { + lifecycleErr <- bridge.SendLifecycleEvent(lifecycleCtx, EventV1{ + Kind: EventDiagnosticV1, Diagnostic: &DiagnosticV1{Code: "test", Message: "test diagnostic"}, + }) + }() + waitForLifecycleEventAdmissionV1(t, bridge) + cancelLifecycle() + if err := <-lifecycleErr; !errors.Is(err, context.Canceled) { + t.Fatalf("SendLifecycleEvent() = %v", err) + } + bridge.eventWrite.mu.Lock() + waiting := bridge.eventWrite.lifecycleWaiting + bridge.eventWrite.mu.Unlock() + if waiting != 0 { + t.Fatalf("lifecycle waiters after cancellation = %d", waiting) + } + close(transport.releaseFirstWrite) + if err := <-writeErr; err != nil { + t.Fatal(err) + } + result, err := bridge.FinalizeOutput(time.Now().Add(time.Second)) + if err != nil || result.Status.Kind != WorkloadOutputFinalizationDrainedV1 { + t.Fatalf("FinalizeOutput() = %#v, %v", result, err) + } + bridge.StopRequests() +} + +func TestSessionIOBridgeV1DoesNotReuseTransportAfterWriteFailure(t *testing.T) { + transport := newPriorityBridgeTestTransportV1() + transport.firstWriteErr = context.DeadlineExceeded + reader, writer := io.Pipe() + bridge, err := StartSessionIOBridgeV1(transport, reader, func(context.Context, RequestV1) error { return nil }) + if err != nil { + t.Fatal(err) + } + writeErr := make(chan error, 1) + go func() { + _, err := writer.Write([]byte("partial output frame")) + writeErr <- errors.Join(err, writer.Close()) + }() + select { + case <-transport.firstWriteEntered: + case <-time.After(time.Second): + t.Fatal("output event was not admitted") + } + close(transport.releaseFirstWrite) + if err := <-writeErr; err != nil && !errors.Is(err, io.ErrClosedPipe) { + t.Fatal(err) + } + result, err := bridge.FinalizeOutput(time.Now().Add(time.Second)) + if err != nil || result.Status.Kind != WorkloadOutputFinalizationFailedV1 || !errors.Is(result.Err, context.DeadlineExceeded) { + t.Fatalf("FinalizeOutput() = %#v, %v", result, err) + } + + lifecycleErr := bridge.SendLifecycleEvent(t.Context(), EventV1{ + Kind: EventDiagnosticV1, Diagnostic: &DiagnosticV1{Code: "test", Message: "must not be written"}, + }) + if lifecycleErr == nil || !strings.Contains(lifecycleErr.Error(), "transport is unusable") { + t.Fatalf("SendLifecycleEvent() = %v", lifecycleErr) + } + transport.mu.Lock() + writes := transport.writeCalls + transport.mu.Unlock() + if writes != 1 { + t.Fatalf("transport event writes after terminal failure = %d, want 1", writes) + } + bridge.StopRequests() +} + +func waitForLifecycleEventAdmissionV1(t *testing.T, bridge *SessionIOBridgeV1) { + t.Helper() + deadline := time.Now().Add(time.Second) + for { + bridge.eventWrite.mu.Lock() + waiting := bridge.eventWrite.lifecycleWaiting + bridge.eventWrite.mu.Unlock() + if waiting == 1 { + return + } + if time.Now().After(deadline) { + t.Fatal("lifecycle event did not enter its bounded admission path") + } + time.Sleep(time.Millisecond) + } +} + +func TestSessionIOBridgeV1SendsOnlyLifecycleEventsAndValidatesConfiguration(t *testing.T) { + transport := newBridgeTestTransportV1() + if _, err := StartSessionIOBridgeV1(nil, io.NopCloser(bytes.NewReader(nil)), func(context.Context, RequestV1) error { return nil }); err == nil { + t.Fatal("StartSessionIOBridgeV1() accepted a missing transport") + } + if _, err := StartSessionIOBridgeV1(transport, io.NopCloser(bytes.NewReader(nil)), nil); err == nil { + t.Fatal("StartSessionIOBridgeV1() accepted a missing handler") + } + if _, err := StartSessionIOBridgeV1(transport, nil, func(context.Context, RequestV1) error { return nil }); err == nil { + t.Fatal("StartSessionIOBridgeV1() accepted a missing output source") + } + + bridge, err := StartSessionIOBridgeV1(transport, io.NopCloser(bytes.NewReader(nil)), func(context.Context, RequestV1) error { return nil }) + if err != nil { + t.Fatal(err) + } + event := EventV1{Kind: EventDiagnosticV1, Diagnostic: &DiagnosticV1{Code: "test", Message: "test diagnostic"}} + if err := bridge.SendLifecycleEvent(t.Context(), event); err != nil { + t.Fatal(err) + } + if got := <-transport.events; !reflect.DeepEqual(got, event) { + t.Fatalf("event = %#v", got) + } + for _, reserved := range []EventV1{ + {Kind: EventOutputV1, Bytes: []byte("forged")}, + {Kind: EventOpenedV1, Opened: &OpenedV1{}}, + } { + if err := bridge.SendLifecycleEvent(t.Context(), reserved); err == nil { + t.Fatalf("SendLifecycleEvent() accepted reserved event %q", reserved.Kind) + } + } + if err := bridge.SendLifecycleEvent(t.Context(), EventV1{ + Kind: EventDiagnosticV1, Diagnostic: &DiagnosticV1{Message: "missing code"}, + }); err == nil { + t.Fatal("SendLifecycleEvent() accepted an invalid lifecycle payload") + } + if err := bridge.WaitRequests(context.Background()); err == nil { + t.Fatal("WaitRequests() accepted a non-cancelable context") + } + bridge.StopRequests() +} + +type bridgeTestRequestResultV1 struct { + request RequestV1 + err error +} + +type bridgeTestTransportV1 struct { + requests chan bridgeTestRequestResultV1 + events chan EventV1 + writeEntered chan struct{} + blockWrites bool +} + +type priorityBridgeTestTransportV1 struct { + firstWriteEntered chan struct{} + releaseFirstWrite chan struct{} + firstWriteOnce sync.Once + firstWriteErr error + + mu sync.Mutex + events []EventV1 + writeCalls int +} + +func newPriorityBridgeTestTransportV1() *priorityBridgeTestTransportV1 { + return &priorityBridgeTestTransportV1{ + firstWriteEntered: make(chan struct{}), + releaseFirstWrite: make(chan struct{}), + } +} + +func (transport *priorityBridgeTestTransportV1) ReadRequest(ctx context.Context) (RequestV1, error) { + <-ctx.Done() + return RequestV1{}, ctx.Err() +} + +func (transport *priorityBridgeTestTransportV1) WriteEvent(ctx context.Context, event EventV1) error { + transport.mu.Lock() + transport.writeCalls++ + transport.mu.Unlock() + first := false + transport.firstWriteOnce.Do(func() { + first = true + close(transport.firstWriteEntered) + }) + if first { + select { + case <-transport.releaseFirstWrite: + case <-ctx.Done(): + return ctx.Err() + } + if transport.firstWriteErr != nil { + return transport.firstWriteErr + } + } + clone := event + clone.Bytes = append([]byte(nil), event.Bytes...) + transport.mu.Lock() + transport.events = append(transport.events, clone) + transport.mu.Unlock() + return nil +} + +func newBridgeTestTransportV1() *bridgeTestTransportV1 { + return &bridgeTestTransportV1{ + requests: make(chan bridgeTestRequestResultV1, 16), + events: make(chan EventV1, 16), + } +} + +func (transport *bridgeTestTransportV1) ReadRequest(ctx context.Context) (RequestV1, error) { + select { + case result := <-transport.requests: + return result.request, result.err + case <-ctx.Done(): + return RequestV1{}, ctx.Err() + } +} + +func (transport *bridgeTestTransportV1) WriteEvent(ctx context.Context, event EventV1) error { + if transport.blockWrites { + select { + case transport.writeEntered <- struct{}{}: + default: + } + <-ctx.Done() + return ctx.Err() + } + clone := event + clone.Bytes = append([]byte(nil), event.Bytes...) + select { + case transport.events <- clone: + return nil + case <-ctx.Done(): + return ctx.Err() + } +} + +type recordingWorkloadPTYControlV1 struct { + mu sync.Mutex + input []byte + columns uint32 + rows uint32 + inputErr error + resizeErr error +} + +func (workload *recordingWorkloadPTYControlV1) WriteInput(_ context.Context, data []byte) error { + workload.mu.Lock() + defer workload.mu.Unlock() + workload.input = append(workload.input, data...) + return workload.inputErr +} + +func (workload *recordingWorkloadPTYControlV1) Resize(_ context.Context, columns uint32, rows uint32) error { + workload.mu.Lock() + defer workload.mu.Unlock() + workload.columns = columns + workload.rows = rows + return workload.resizeErr +} diff --git a/internal/dockerdeploy/controlled_session_session_io_integration_test.go b/internal/dockerdeploy/controlled_session_session_io_integration_test.go new file mode 100644 index 0000000..782dd20 --- /dev/null +++ b/internal/dockerdeploy/controlled_session_session_io_integration_test.go @@ -0,0 +1,317 @@ +package dockerdeploy + +import ( + "bytes" + "context" + "errors" + "fmt" + "net" + "os" + "os/exec" + "path/filepath" + "runtime" + "strconv" + "strings" + "sync" + "testing" + "time" + + "github.com/omry/reploy/internal/controlledsession" +) + +func TestControlledSessionIOBridgeDockerIntegration(t *testing.T) { + if os.Getenv("REPLOY_DOCKER_INTEGRATION") != "1" { + t.Skip("set REPLOY_DOCKER_INTEGRATION=1 to run Docker integration evidence") + } + if runtime.GOOS != "linux" || runtime.GOARCH != "amd64" && runtime.GOARCH != "arm64" { + t.Skipf("controlled-session I/O bridge integration requires a supported Linux host, got %s/%s", runtime.GOOS, runtime.GOARCH) + } + ctx, cancel := context.WithTimeout(context.Background(), 4*time.Minute) + defer cancel() + image, _ := buildApplicationStartupVerifierIntegrationImage(t, ctx) + + t.Run("typed requests and ordered PTY output", func(t *testing.T) { + workload, bridge, client, requests := prepareControlledSessionIOBridgeIntegrationV1(t, ctx, image) + capture := startSessionIOEventCaptureV1(client) + if err := workload.Start(ctx); err != nil { + t.Fatal(err) + } + + writeSessionIORequestIntegrationV1(t, client, controlledsession.RequestV1{ + Kind: controlledsession.RequestInputV1, + Bytes: []byte("stty size; printf 'SIZE-1-DONE\\n'\n"), + }) + waitSessionIORequestIntegrationV1(t, requests, controlledsession.RequestInputV1) + capture.waitForOutput(t, []byte("24 80"), 10*time.Second) + capture.waitForOutput(t, []byte("SIZE-1-DONE"), 10*time.Second) + + writeSessionIORequestIntegrationV1(t, client, controlledsession.RequestV1{ + Kind: controlledsession.RequestResizeV1, Columns: 132, Rows: 43, + }) + waitSessionIORequestIntegrationV1(t, requests, controlledsession.RequestResizeV1) + writeSessionIORequestIntegrationV1(t, client, controlledsession.RequestV1{ + Kind: controlledsession.RequestInputV1, + Bytes: []byte("stty size; printf '\\001\\002\\177\\377'; printf 'SIZE-2-DONE\\n'\n"), + }) + waitSessionIORequestIntegrationV1(t, requests, controlledsession.RequestInputV1) + capture.waitForOutput(t, []byte("43 132"), 10*time.Second) + capture.waitForOutput(t, []byte{0x01, 0x02, 0x7f, 0xff}, 10*time.Second) + capture.waitForOutput(t, []byte("SIZE-2-DONE"), 10*time.Second) + + writeSessionIORequestIntegrationV1(t, client, controlledsession.RequestV1{ + Kind: controlledsession.RequestInputV1, + Bytes: []byte("printf '\\036SLEEP-ACTIVE\\n'; sleep 30\n"), + }) + waitSessionIORequestIntegrationV1(t, requests, controlledsession.RequestInputV1) + capture.waitForOutput(t, append([]byte{0x1e}, []byte("SLEEP-ACTIVE")...), 10*time.Second) + writeSessionIORequestIntegrationV1(t, client, controlledsession.RequestV1{ + Kind: controlledsession.RequestInputV1, Bytes: []byte{0x03}, + }) + waitSessionIORequestIntegrationV1(t, requests, controlledsession.RequestInputV1) + writeSessionIORequestIntegrationV1(t, client, controlledsession.RequestV1{ + Kind: controlledsession.RequestInputV1, + Bytes: []byte("printf 'INTERRUPT-DONE\\n'; exit 42\n"), + }) + waitSessionIORequestIntegrationV1(t, requests, controlledsession.RequestInputV1) + capture.waitForOutput(t, []byte("INTERRUPT-DONE"), 10*time.Second) + + status, err := workload.Wait(ctx) + if err != nil { + t.Fatal(err) + } + if status.Kind != controlledsession.ProcessStatusExitedV1 || status.Code == nil || *status.Code != 42 { + t.Fatalf("workload status = %#v", status) + } + result, err := bridge.FinalizeOutput(time.Now().Add(10 * time.Second)) + if err != nil || result.Status.Kind != controlledsession.WorkloadOutputFinalizationDrainedV1 || result.Err != nil { + t.Fatalf("FinalizeOutput() = %#v, %v", result, err) + } + bridge.StopRequests() + if err := bridge.WaitRequests(ctx); err != nil { + t.Fatalf("WaitRequests() = %v", err) + } + if err := client.Close(); err != nil { + t.Fatal(err) + } + capture.waitDone(t, 10*time.Second) + }) + + t.Run("slow output preserves request flow and disconnect fails delivery", func(t *testing.T) { + workload, bridge, client, requests := prepareControlledSessionIOBridgeIntegrationV1(t, ctx, image) + if err := workload.Start(ctx); err != nil { + t.Fatal(err) + } + writeSessionIORequestIntegrationV1(t, client, controlledsession.RequestV1{ + Kind: controlledsession.RequestInputV1, + Bytes: []byte("head -c 16777216 /dev/zero; printf 'SLOW-DONE\\n'\n"), + }) + waitSessionIORequestIntegrationV1(t, requests, controlledsession.RequestInputV1) + time.Sleep(200 * time.Millisecond) + writeSessionIORequestIntegrationV1(t, client, controlledsession.RequestV1{ + Kind: controlledsession.RequestResizeV1, Columns: 101, Rows: 37, + }) + waitSessionIORequestIntegrationV1(t, requests, controlledsession.RequestResizeV1) + select { + case <-bridge.OutputDone(): + t.Fatal("output bridge stopped instead of applying backpressure") + default: + } + if err := client.Close(); err != nil { + t.Fatal(err) + } + if err := bridge.WaitRequests(ctx); !errors.Is(err, controlledsession.ErrControllerDisconnectedV1) { + t.Fatalf("WaitRequests() = %v", err) + } + result, err := bridge.FinalizeOutput(time.Now().Add(10 * time.Second)) + if err != nil || result.Status.Kind != controlledsession.WorkloadOutputFinalizationFailedV1 || + !errors.Is(result.Err, controlledsession.ErrControllerDisconnectedV1) { + t.Fatalf("FinalizeOutput() = %#v, %v", result, err) + } + }) +} + +func prepareControlledSessionIOBridgeIntegrationV1( + t *testing.T, + ctx context.Context, + image string, +) (*DockerWorkloadPTYV1, *controlledsession.SessionIOBridgeV1, *net.UnixConn, <-chan controlledsession.RequestV1) { + t.Helper() + identity := controlledsession.RuntimeIdentityV1{ + Username: "reploy", UID: strconv.Itoa(os.Geteuid()), GID: strconv.Itoa(os.Getegid()), SupplementaryGIDs: []string{}, + } + if os.Geteuid() == 0 { + identity.Username = "root" + } + authorization := testControlledSessionChannelAuthorizationV1(t, identity) + channel, err := controlledsession.PreparePrivateChannelV1(controlledsession.PrivateChannelConfigV1{ + HostDirectory: filepath.Join(shortControlledSessionChannelTestDirectoryV1(t), "bridge"), + Opened: controlledsession.OpenedV1{ + Authorization: authorization, Columns: 80, Rows: 24, + OutputFinalizationTimeoutMilliseconds: controlledsession.DefaultOutputFinalizationTimeoutMillisecondsV1, + }, + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = channel.Close() }) + claim := make(chan struct { + connection *controlledsession.ControllerConnectionV1 + err error + }, 1) + go func() { + connection, err := channel.Claim(ctx) + claim <- struct { + connection *controlledsession.ControllerConnectionV1 + err error + }{connection: connection, err: err} + }() + client, err := net.DialUnix("unix", nil, &net.UnixAddr{Name: channel.SocketPath(), Net: "unix"}) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = client.Close() }) + if opened, err := controlledsession.ReadEventV1(client); err != nil || opened.Kind != controlledsession.EventOpenedV1 { + t.Fatalf("opened event = %#v, %v", opened, err) + } + claimed := <-claim + if claimed.err != nil { + t.Fatal(claimed.err) + } + + plan := controlledSessionWorkloadIntegrationPlanV1(t, image, 80, 24) + workload, err := PrepareDockerWorkloadPTYV1(ctx, plan) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { + _ = workload.Close() + cleanupCtx, cancel := context.WithTimeout(context.Background(), 10*time.Second) + defer cancel() + output, err := exec.CommandContext(cleanupCtx, plan.Cleanup.Name, plan.Cleanup.Args...).CombinedOutput() + if err != nil && !strings.Contains(string(output), "No such container") { + t.Errorf("cleanup controlled-session I/O bridge workload %q: %v\n%s", plan.Container, err, output) + } + }) + output, err := workload.Output() + if err != nil { + t.Fatal(err) + } + machine, err := controlledsession.NewMachineV1(authorization) + if err != nil { + t.Fatal(err) + } + if _, err := machine.Observe(controlledsession.ObservationV1{Kind: controlledsession.ObservationActivatedV1}); err != nil { + t.Fatal(err) + } + requests := make(chan controlledsession.RequestV1, 16) + bridge, err := controlledsession.StartSessionIOBridgeV1(claimed.connection, output, func(requestCtx context.Context, request controlledsession.RequestV1) error { + if _, err := machine.ApplyRequest(request); err != nil { + return err + } + handled, err := controlledsession.ApplyAcceptedWorkloadPTYRequestV1(requestCtx, workload, request) + if err != nil { + return err + } + if !handled { + return fmt.Errorf("integration request %q unexpectedly has no PTY effect", request.Kind) + } + requests <- request + return nil + }) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { bridge.StopRequests() }) + return workload, bridge, client, requests +} + +func writeSessionIORequestIntegrationV1(t *testing.T, client *net.UnixConn, request controlledsession.RequestV1) { + t.Helper() + if err := controlledsession.WriteRequestV1(client, request); err != nil { + t.Fatal(err) + } +} + +func waitSessionIORequestIntegrationV1(t *testing.T, requests <-chan controlledsession.RequestV1, want controlledsession.RequestKindV1) { + t.Helper() + select { + case request := <-requests: + if request.Kind != want { + t.Fatalf("handled request = %#v, want kind %q", request, want) + } + case <-time.After(10 * time.Second): + t.Fatalf("timed out waiting for handled %q request", want) + } +} + +type sessionIOEventCaptureV1 struct { + mu sync.Mutex + output []byte + done chan struct{} + notify chan struct{} + err error +} + +func startSessionIOEventCaptureV1(client *net.UnixConn) *sessionIOEventCaptureV1 { + capture := &sessionIOEventCaptureV1{done: make(chan struct{}), notify: make(chan struct{}, 1)} + go func() { + defer close(capture.done) + for { + event, err := controlledsession.ReadEventV1(client) + capture.mu.Lock() + if err != nil { + if !errors.Is(err, net.ErrClosed) { + capture.err = err + } + capture.mu.Unlock() + return + } + if event.Kind == controlledsession.EventOutputV1 { + capture.output = append(capture.output, event.Bytes...) + } + capture.mu.Unlock() + select { + case capture.notify <- struct{}{}: + default: + } + } + }() + return capture +} + +func (capture *sessionIOEventCaptureV1) waitForOutput(t *testing.T, want []byte, timeout time.Duration) { + t.Helper() + timer := time.NewTimer(timeout) + defer timer.Stop() + for { + capture.mu.Lock() + found := bytes.Contains(capture.output, want) + output := append([]byte(nil), capture.output...) + err := capture.err + capture.mu.Unlock() + if found { + return + } + select { + case <-capture.notify: + case <-capture.done: + t.Fatalf("controller event stream closed before %q; output = %q, error = %v", want, output, err) + case <-timer.C: + t.Fatalf("timed out waiting for controller output %q; output = %q", want, output) + } + } +} + +func (capture *sessionIOEventCaptureV1) waitDone(t *testing.T, timeout time.Duration) { + t.Helper() + select { + case <-capture.done: + capture.mu.Lock() + defer capture.mu.Unlock() + if capture.err != nil { + t.Fatalf("controller event capture failed: %v", capture.err) + } + case <-time.After(timeout): + t.Fatal("timed out waiting for controller event capture to stop") + } +}