From d05ab95d569df763d1dbae537835ca70fdd12f67 Mon Sep 17 00:00:00 2001 From: Omry Yadan Date: Sun, 9 Aug 2026 05:41:58 +0800 Subject: [PATCH] Supervise controlled-session lifecycle --- docs/CONTROLLED_SESSION_DESIGN.md | 43 +- internal/controlledsession/lifecycle.go | 52 +- internal/controlledsession/lifecycle_test.go | 129 ++- internal/controlledsession/output.go | 32 + internal/controlledsession/output_test.go | 12 + internal/controlledsession/session_io.go | 45 +- internal/controlledsession/session_io_test.go | 40 + .../controlled_session_controller.go | 4 +- ...led_session_controller_integration_test.go | 6 +- .../controlled_session_controller_test.go | 4 +- .../controlled_session_supervisor.go | 870 ++++++++++++++++ ...led_session_supervisor_integration_test.go | 56 ++ .../controlled_session_supervisor_test.go | 938 ++++++++++++++++++ .../controlled_session_workload_pty.go | 83 +- .../controlled_session_workload_pty_test.go | 102 +- .../testdata/session_channel_helper/main.go | 88 +- 16 files changed, 2443 insertions(+), 61 deletions(-) create mode 100644 internal/dockerdeploy/controlled_session_supervisor.go create mode 100644 internal/dockerdeploy/controlled_session_supervisor_integration_test.go create mode 100644 internal/dockerdeploy/controlled_session_supervisor_test.go diff --git a/docs/CONTROLLED_SESSION_DESIGN.md b/docs/CONTROLLED_SESSION_DESIGN.md index dad61a3..b1c944b 100644 --- a/docs/CONTROLLED_SESSION_DESIGN.md +++ b/docs/CONTROLLED_SESSION_DESIGN.md @@ -46,8 +46,17 @@ summary: Capability-scoped execution sessions that inherit Reploy's global conta 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. + to a potentially partial frame. The attached host lifecycle supervisor is + also implemented: it prepares inert resources, starts the controller before + the workload, activates only after workload setup succeeds, serializes + controller requests through the lifecycle machine, latches the first + termination cause, stops and independently observes the workload, finalizes + PTY output before the terminal result, waits boundedly for controller + completion and result acknowledgement, and removes both containers and the + private channel. A workload that starts before a later startup step fails is + still terminated and its output is finalized through the same barrier. + Crash watchdogs and restart reconciliation remain the next ownership phase; + controlled-session networking remains a later phase. - Initial runtime: Linux containers under Docker - Motivating clients: OmegaFlow recording, sandboxed AI agents, security inspection, and untrusted-code execution @@ -468,8 +477,8 @@ bytes are never parsed as protocol messages. - `terminate`: request bounded graceful session termination. - `complete`: after Host Reploy has emitted `workload_outputs_finalized`, declare that the controller has finalized its client-owned results. It does not stop - an active workload and is rejected before workload output reaches a terminal - state. + an active workload and is rejected until successful publication of the + workload-output-finalization event. - `acknowledge_terminated`: confirm receipt of the authoritative `terminated` event. This payload-free protocol handshake is mandatory housekeeping, not a granted capability, and is accepted only after Host Reploy has successfully @@ -1071,27 +1080,33 @@ OmegaFlow these include the recording artifacts. Host Reploy does not open a controller-finalization wait when `complete` was not granted and records that controller as `not-completed` in the terminal result. Repeated terminate or host cancel operations are idempotent. Input and resize -are rejected after `terminating` begins. A single `complete` remains valid -during termination while Host Reploy is waiting for controller finalization. A -`failed` workload-output result makes the session fail regardless of whether -the controller preserves and finalizes partial artifacts. +are rejected after `terminating` begins, and Host Reploy cancels any accepted +input or resize operation still blocked in the runtime before it begins +workload teardown. This cancellation does not stop request dispatch: a single +`complete` remains valid during termination while Host Reploy is waiting for +controller finalization. A `failed` workload-output result makes the session +fail regardless of whether the controller preserves and finalizes partial +artifacts. Normal completion is: 1. Host Reploy observes workload exit, or a controller or host operation requests termination. 2. Host Reploy atomically latches the cause and enters `terminating`. -3. Host Reploy performs bounded graceful termination followed by forced - termination when necessary. +3. Host Reploy cancels any in-flight workload request, then performs bounded + graceful termination followed by forced termination when necessary. 4. Host Reploy independently observes the exact workload container stopped. 5. Host Reploy drains and closes every declared workload-output surface under the finite output-finalization deadline, then emits the one ordered `workload_outputs_finalized` outcome. 6. When the live controller was granted `complete`, Host Reploy gives it a - bounded finalization period in which to close its client-owned output and - send `complete`. Without that grant, Host Reploy skips the wait and records - `not-completed`. A failed output outcome remains a session failure even when - partial client artifacts are finalized. + bounded finalization period only after that event is successfully published. + A response arriving while publication finishes is held until publication's + authoritative outcome is recorded; an earlier response is rejected. The + controller may then close its client-owned output and send `complete`. + Without that grant, Host Reploy skips the wait and records `not-completed`. + A failed output outcome remains a session failure even when partial client + artifacts are finalized. 7. Host Reploy removes the workload container, temporary mounts, networks, and every other lease resource not required to deliver the final result. It keeps the controller and private session channel alive. diff --git a/internal/controlledsession/lifecycle.go b/internal/controlledsession/lifecycle.go index 72e832a..d04953b 100644 --- a/internal/controlledsession/lifecycle.go +++ b/internal/controlledsession/lifecycle.go @@ -33,6 +33,7 @@ const ( ObservationRuntimeObservationLostV1 ObservationKindV1 = "runtime-observation-lost" ObservationStartupFailureV1 ObservationKindV1 = "startup-failure" ObservationWorkloadOutputsFinalizedV1 ObservationKindV1 = "workload-outputs-finalized" + ObservationWorkloadOutputsPublishedV1 ObservationKindV1 = "workload-outputs-published" ObservationWorkloadOutputFinalizationExpiredV1 ObservationKindV1 = "workload-output-finalization-expired" ObservationControllerFinalizationExpiredV1 ObservationKindV1 = "controller-finalization-expired" ObservationFinishedV1 ObservationKindV1 = "finished" @@ -43,8 +44,13 @@ type ObservationV1 struct { Kind ObservationKindV1 WorkloadStatus *ProcessStatusV1 WorkloadOutputFinalizationStatus *WorkloadOutputFinalizationStatusV1 - Reason string - Finish *FinishV1 + // WorkloadOutputPending records the narrow startup-failure case where + // the runtime started the workload before a later startup operation failed. + // The supervisor must then stop the workload and explicitly finalize its + // output before the lifecycle can finish. + WorkloadOutputPending bool + Reason string + Finish *FinishV1 } type SnapshotV1 struct { @@ -55,6 +61,7 @@ type SnapshotV1 struct { RuntimeObservationStatus RuntimeObservationStatusV1 ControllerFinalizationStatus ControllerFinalizationStatusV1 AwaitingWorkloadOutputFinalization bool + AwaitingWorkloadOutputPublication bool AwaitingControllerFinalization bool AwaitingResultAcknowledgement bool ResultAcknowledged bool @@ -69,6 +76,7 @@ type TransitionV1 struct { BeginTermination bool WorkloadOutputFinalizationStatus WorkloadOutputFinalizationStatusV1 AwaitingWorkloadOutputFinalization bool + AwaitingWorkloadOutputPublication bool AwaitingControllerFinalization bool AwaitingResultAcknowledgement bool RequestAccepted bool @@ -93,6 +101,7 @@ type MachineV1 struct { controller ControllerFinalizationStatusV1 runtimeObservation RuntimeObservationStatusV1 waitingOutputs bool + waitingPublication bool waitingFinalize bool resultDelivered bool resultAcknowledged bool @@ -123,6 +132,9 @@ func (machine *MachineV1) Observe(observation ObservationV1) (TransitionV1, erro defer machine.mu.Unlock() before := machine.state transition := TransitionV1{Before: before, After: before, Cause: machine.cause} + if observation.WorkloadOutputPending && observation.Kind != ObservationStartupFailureV1 { + return transition, fmt.Errorf("%w: pending workload output is valid only for startup failure", ErrObservationRejected) + } if observation.Kind == ObservationResultDeliveredV1 { if observation.WorkloadStatus != nil || observation.WorkloadOutputFinalizationStatus != nil || observation.Reason != "" || observation.Finish != nil || machine.state != StateTerminatedV1 || machine.result == nil { return transition, fmt.Errorf("%w: result delivery is valid only after termination and carries no payload", ErrObservationRejected) @@ -184,6 +196,7 @@ func (machine *MachineV1) Observe(observation ObservationV1) (TransitionV1, erro machine.controller = ControllerFinalizationStatusV1{Kind: ControllerFinalizationLostV1, Reason: observation.Reason} } machine.waitingFinalize = false + machine.waitingPublication = false machine.latchLocked(CauseControllerLostV1, &transition) case ObservationRuntimeObservationLostV1: if err := validateCauseObservationV1(observation); err != nil { @@ -198,17 +211,29 @@ func (machine *MachineV1) Observe(observation ObservationV1) (TransitionV1, erro // reports failed closure or its bounded finalization deadline expires. machine.finalizePreActivationOutputsForRuntimeObservationLossLocked(observation.Reason) case ObservationStartupFailureV1: - if observation.WorkloadStatus != nil || observation.WorkloadOutputFinalizationStatus != nil || observation.Finish != nil { - return transition, fmt.Errorf("%w: startup failure carries only a reason", ErrObservationRejected) + if observation.WorkloadStatus != nil || observation.Finish != nil || + (observation.WorkloadOutputPending && observation.WorkloadOutputFinalizationStatus != nil) { + return transition, fmt.Errorf("%w: startup failure carries a reason and at most one workload-output outcome", ErrObservationRejected) } if err := validateRequiredSafeTextV1("startup-failure reason", observation.Reason); err != nil { return transition, fmt.Errorf("%w: %v", ErrObservationRejected, err) } + if observation.WorkloadOutputFinalizationStatus != nil { + if err := validateWorkloadOutputFinalizationStatusV1(*observation.WorkloadOutputFinalizationStatus); err != nil { + return transition, fmt.Errorf("%w: %v", ErrObservationRejected, err) + } + } if machine.controller.Kind != ControllerFinalizationUnknownV1 { return transition, fmt.Errorf("%w: startup failure is invalid after controller activation", ErrObservationRejected) } machine.controller = ControllerFinalizationStatusV1{Kind: ControllerFinalizationStartupFailedV1, Reason: observation.Reason} machine.latchLocked(CauseStartupFailureV1, &transition) + if observation.WorkloadOutputPending { + machine.workloadOutputs = WorkloadOutputFinalizationStatusV1{} + machine.waitingOutputs = true + } else if observation.WorkloadOutputFinalizationStatus != nil { + machine.completeOutputFinalizationLocked(*observation.WorkloadOutputFinalizationStatus) + } case ObservationWorkloadOutputsFinalizedV1: if observation.WorkloadStatus != nil || observation.WorkloadOutputFinalizationStatus == nil || observation.Reason != "" || observation.Finish != nil || !machine.waitingOutputs { return transition, fmt.Errorf("%w: workload output finalization requires exactly one status while output finalization is pending", ErrObservationRejected) @@ -225,6 +250,14 @@ func (machine *MachineV1) Observe(observation ObservationV1) (TransitionV1, erro return transition, fmt.Errorf("%w: runtime observation loss requires failed workload output finalization", ErrObservationRejected) } machine.completeOutputFinalizationLocked(*observation.WorkloadOutputFinalizationStatus) + case ObservationWorkloadOutputsPublishedV1: + if observation.WorkloadStatus != nil || observation.WorkloadOutputFinalizationStatus != nil || observation.Reason != "" || observation.Finish != nil || + machine.state != StateTerminatingV1 || machine.waitingOutputs || machine.workloadOutputs.Kind == "" || !machine.waitingPublication { + return transition, fmt.Errorf("%w: workload output publication is valid only after output finalization and carries no payload", ErrObservationRejected) + } + machine.waitingPublication = false + machine.waitingFinalize = machine.controller.Kind == ControllerFinalizationActiveV1 && + containsOperationV1(machine.authorization.Operations, OperationCompleteV1) case ObservationWorkloadOutputFinalizationExpiredV1: if observation.WorkloadStatus != nil || observation.WorkloadOutputFinalizationStatus != nil || observation.Finish != nil || !machine.waitingOutputs { return transition, fmt.Errorf("%w: output-finalization expiry carries only a required reason while output finalization is pending", ErrObservationRejected) @@ -246,8 +279,8 @@ func (machine *MachineV1) Observe(observation ObservationV1) (TransitionV1, erro if observation.WorkloadStatus != nil || observation.WorkloadOutputFinalizationStatus != nil || observation.Reason != "" || observation.Finish == nil { return transition, fmt.Errorf("%w: finish requires exactly one terminal status set", ErrObservationRejected) } - if machine.state != StateTerminatingV1 || machine.waitingOutputs || machine.workloadOutputs.Kind == "" || machine.waitingFinalize { - return transition, fmt.Errorf("%w: finish requires finalized workload output and no pending output or controller finalization", ErrObservationRejected) + if machine.state != StateTerminatingV1 || machine.waitingOutputs || machine.waitingPublication || machine.workloadOutputs.Kind == "" || machine.waitingFinalize { + return transition, fmt.Errorf("%w: finish requires published workload output finalization and no pending output or controller finalization", ErrObservationRejected) } if err := validateFinishV1(*observation.Finish); err != nil { return transition, fmt.Errorf("%w: %v", ErrObservationRejected, err) @@ -283,6 +316,7 @@ func (machine *MachineV1) Observe(observation ObservationV1) (TransitionV1, erro transition.Cause = machine.cause transition.WorkloadOutputFinalizationStatus = machine.workloadOutputs transition.AwaitingWorkloadOutputFinalization = machine.waitingOutputs + transition.AwaitingWorkloadOutputPublication = machine.waitingPublication transition.AwaitingControllerFinalization = machine.waitingFinalize transition.AwaitingResultAcknowledgement = machine.resultDelivered && !machine.resultAcknowledged transition.ResultAcknowledged = machine.resultAcknowledged @@ -364,6 +398,7 @@ func (machine *MachineV1) ApplyRequest(request RequestV1) (TransitionV1, error) transition.Cause = machine.cause transition.WorkloadOutputFinalizationStatus = machine.workloadOutputs transition.AwaitingWorkloadOutputFinalization = machine.waitingOutputs + transition.AwaitingWorkloadOutputPublication = machine.waitingPublication transition.AwaitingControllerFinalization = machine.waitingFinalize transition.AwaitingResultAcknowledgement = machine.resultDelivered && !machine.resultAcknowledged transition.RequestAccepted = true @@ -395,6 +430,7 @@ func (machine *MachineV1) snapshotLocked() SnapshotV1 { RuntimeObservationStatus: machine.runtimeObservation, ControllerFinalizationStatus: machine.controller, AwaitingControllerFinalization: machine.waitingFinalize, AwaitingWorkloadOutputFinalization: machine.waitingOutputs, + AwaitingWorkloadOutputPublication: machine.waitingPublication, AwaitingResultAcknowledgement: machine.resultDelivered && !machine.resultAcknowledged, ResultAcknowledged: machine.resultAcknowledged, Result: cloneResultV1(machine.result), } @@ -441,8 +477,8 @@ func (machine *MachineV1) finalizePreActivationOutputsForRuntimeObservationLossL func (machine *MachineV1) completeOutputFinalizationLocked(status WorkloadOutputFinalizationStatusV1) { machine.workloadOutputs = status machine.waitingOutputs = false - machine.waitingFinalize = machine.controller.Kind == ControllerFinalizationActiveV1 && - containsOperationV1(machine.authorization.Operations, OperationCompleteV1) + machine.waitingPublication = machine.controller.Kind == ControllerFinalizationActiveV1 + machine.waitingFinalize = false } func equalProcessStatusV1(left ProcessStatusV1, right ProcessStatusV1) bool { diff --git a/internal/controlledsession/lifecycle_test.go b/internal/controlledsession/lifecycle_test.go index bc10d40..6233750 100644 --- a/internal/controlledsession/lifecycle_test.go +++ b/internal/controlledsession/lifecycle_test.go @@ -41,6 +41,18 @@ func observeOutputsFinalizedV1(t *testing.T, machine *MachineV1, status Workload if err != nil { t.Fatalf("Observe(workload outputs finalized) error = %v", err) } + if machine.Snapshot().ControllerFinalizationStatus.Kind != ControllerFinalizationActiveV1 { + return transition + } + return observeOutputsPublishedV1(t, machine) +} + +func observeOutputsPublishedV1(t *testing.T, machine *MachineV1) TransitionV1 { + t.Helper() + transition, err := machine.Observe(ObservationV1{Kind: ObservationWorkloadOutputsPublishedV1}) + if err != nil { + t.Fatalf("Observe(workload outputs published) error = %v", err) + } return transition } @@ -105,6 +117,45 @@ func TestLifecycleOutputBarrierPrecedesControllerFinalization(t *testing.T) { } } +func TestLifecycleRejectsCompletionUntilOutputFinalizationIsPublished(t *testing.T) { + machine := activatedMachineV1(t) + code := 0 + observeWorkloadExitV1(t, machine, code) + outputStatus := WorkloadOutputFinalizationStatusV1{Kind: WorkloadOutputFinalizationDrainedV1} + finalized, err := machine.Observe(ObservationV1{ + Kind: ObservationWorkloadOutputsFinalizedV1, + WorkloadOutputFinalizationStatus: &outputStatus, + }) + if err != nil { + t.Fatal(err) + } + if finalized.AwaitingWorkloadOutputFinalization || !finalized.AwaitingWorkloadOutputPublication || finalized.AwaitingControllerFinalization { + t.Fatalf("internal output finalization opened controller finalization early: %#v", finalized) + } + if _, err := machine.ApplyRequest(RequestV1{Kind: RequestCompleteV1}); !errors.Is(err, ErrRequestRejected) { + t.Fatalf("complete before output publication error = %v", err) + } + if _, err := machine.Observe(ObservationV1{Kind: ObservationFinishedV1, Finish: &FinishV1{ + WorkloadStatus: ProcessStatusV1{Kind: ProcessStatusExitedV1, Code: &code}, + WorkloadOutputFinalizationStatus: outputStatus, + ControllerFinalizationStatus: ControllerFinalizationStatusV1{Kind: ControllerFinalizationNotCompletedV1}, + CleanupStatus: CleanupStatusV1{Kind: CleanupStatusSucceededV1}, + RecoveryAction: RecoveryNoneV1, + }}); !errors.Is(err, ErrObservationRejected) { + t.Fatalf("finish before output publication error = %v", err) + } + published, err := machine.Observe(ObservationV1{Kind: ObservationWorkloadOutputsPublishedV1}) + if err != nil || published.AwaitingWorkloadOutputPublication || !published.AwaitingControllerFinalization { + t.Fatalf("output publication transition = %#v, %v", published, err) + } + if _, err := machine.Observe(ObservationV1{Kind: ObservationWorkloadOutputsPublishedV1}); !errors.Is(err, ErrObservationRejected) { + t.Fatalf("duplicate output publication error = %v", err) + } + if _, err := machine.ApplyRequest(RequestV1{Kind: RequestCompleteV1}); err != nil { + t.Fatalf("complete after output publication error = %v", err) + } +} + func TestLifecycleSkipsFinalizationWaitWithoutCompleteGrant(t *testing.T) { authorization := testAuthorizationV1() authorization.Operations = []OperationV1{OperationInputV1, OperationResizeV1, OperationTerminateV1} @@ -412,9 +463,13 @@ func TestLifecycleOutputFinalizationExpiryCannotBecomeDrained(t *testing.T) { if err != nil { t.Fatal(err) } - if transition.AwaitingWorkloadOutputFinalization || !transition.AwaitingControllerFinalization { + if transition.AwaitingWorkloadOutputFinalization || transition.AwaitingControllerFinalization { t.Fatalf("output timeout transition = %#v", transition) } + published := observeOutputsPublishedV1(t, machine) + if !published.AwaitingControllerFinalization { + t.Fatalf("output timeout publication transition = %#v", published) + } drained := WorkloadOutputFinalizationStatusV1{Kind: WorkloadOutputFinalizationDrainedV1} if _, err := machine.Observe(ObservationV1{ @@ -505,6 +560,9 @@ func TestLifecycleOutputFinalizationExpiryAcceptsLateWorkloadExit(t *testing.T) }); err != nil { t.Fatalf("late workload exit error = %v", err) } + if machine.Snapshot().ControllerFinalizationStatus.Kind == ControllerFinalizationActiveV1 { + observeOutputsPublishedV1(t, machine) + } snapshot := machine.Snapshot() failed := WorkloadOutputFinalizationStatusV1{ @@ -703,6 +761,75 @@ func TestLifecycleStartupFailureIsLimitedToPreActivation(t *testing.T) { } } +func TestLifecycleStartupFailureCanWaitForPartiallyStartedWorkloadOutput(t *testing.T) { + machine, err := NewMachineV1(testAuthorizationV1()) + if err != nil { + t.Fatal(err) + } + transition, err := machine.Observe(ObservationV1{ + Kind: ObservationStartupFailureV1, + Reason: "initial terminal resize failed", + WorkloadOutputPending: true, + }) + if err != nil { + t.Fatal(err) + } + if transition.Cause != CauseStartupFailureV1 || !transition.AwaitingWorkloadOutputFinalization || + transition.WorkloadOutputFinalizationStatus.Kind != "" { + t.Fatalf("startup transition = %#v", transition) + } + + code := 143 + if _, err := machine.Observe(ObservationV1{ + Kind: ObservationWorkloadExitV1, + WorkloadStatus: &ProcessStatusV1{Kind: ProcessStatusExitedV1, Code: &code}, + }); err != nil { + t.Fatal(err) + } + drained := WorkloadOutputFinalizationStatusV1{Kind: WorkloadOutputFinalizationDrainedV1} + barrier, err := machine.Observe(ObservationV1{ + Kind: ObservationWorkloadOutputsFinalizedV1, + WorkloadOutputFinalizationStatus: &drained, + }) + if err != nil { + t.Fatal(err) + } + if barrier.AwaitingWorkloadOutputFinalization || barrier.AwaitingControllerFinalization { + t.Fatalf("finalization transition = %#v", barrier) + } +} + +func TestLifecycleStartupFailureAcceptsAlreadyFinalizedInertOutput(t *testing.T) { + machine, err := NewMachineV1(testAuthorizationV1()) + if err != nil { + t.Fatal(err) + } + failed := WorkloadOutputFinalizationStatusV1{ + Kind: WorkloadOutputFinalizationFailedV1, + Reason: "workload output closure failed", + } + transition, err := machine.Observe(ObservationV1{ + Kind: ObservationStartupFailureV1, + Reason: "workload start failed", + WorkloadOutputFinalizationStatus: &failed, + }) + if err != nil { + t.Fatal(err) + } + if transition.AwaitingWorkloadOutputFinalization || transition.WorkloadOutputFinalizationStatus != failed { + t.Fatalf("startup transition = %#v", transition) + } + + if _, err := machine.Observe(ObservationV1{ + Kind: ObservationStartupFailureV1, + Reason: "invalid duplicate", + WorkloadOutputPending: true, + WorkloadOutputFinalizationStatus: &failed, + }); !errors.Is(err, ErrObservationRejected) { + t.Fatalf("ambiguous startup output state error = %v", err) + } +} + func TestLifecycleEnforcesAuthorizationAndCopiesIt(t *testing.T) { authorization := testAuthorizationV1() authorization.Operations = []OperationV1{OperationCompleteV1} diff --git a/internal/controlledsession/output.go b/internal/controlledsession/output.go index ccdb1fb..7c28e10 100644 --- a/internal/controlledsession/output.go +++ b/internal/controlledsession/output.go @@ -88,6 +88,38 @@ func (pump *PTYOutputPumpV1) Done() <-chan struct{} { return pump.done } +// TerminalResult returns the raw immutable pump result after Done closes. It +// does not apply the host-owned finalization deadline; lifecycle supervisors +// use it only to distinguish a failed output surface from ordinary drained +// EOF before establishing the termination cause. +func (pump *PTYOutputPumpV1) TerminalResult() (PTYOutputFinalizationV1, bool) { + select { + case <-pump.done: + pump.resultMu.Lock() + defer pump.resultMu.Unlock() + if pump.result == nil { + panic("controlled-session PTY output pump stopped without a terminal result") + } + return *pump.result, true + default: + return PTYOutputFinalizationV1{}, false + } +} + +// IsPTYOutputDeliveryFailureV1 reports whether output failed while being +// delivered to its consumer rather than while reading or closing the PTY. +func IsPTYOutputDeliveryFailureV1(result PTYOutputFinalizationV1) bool { + return result.Status == failedPTYOutputStatusV1(ptyOutputDeliveryFailureReasonV1) +} + +// IsPTYOutputObservationFailureV1 reports whether the host lost the workload +// PTY while reading or closing it. Delivery failure and a host-owned +// finalization timeout have distinct lifecycle meanings. +func IsPTYOutputObservationFailureV1(result PTYOutputFinalizationV1) bool { + return result.Status == failedPTYOutputStatusV1(ptyOutputReadFailureReasonV1) || + result.Status == failedPTYOutputStatusV1(ptyOutputClosureFailureReasonV1) +} + // Finalize waits for all previously read bytes to be delivered and for the // source to close. deadline is the absolute host-owned deadline established // when termination began; time already spent stopping the workload is not diff --git a/internal/controlledsession/output_test.go b/internal/controlledsession/output_test.go index 978c020..86ba0fb 100644 --- a/internal/controlledsession/output_test.go +++ b/internal/controlledsession/output_test.go @@ -123,6 +123,9 @@ func TestPTYOutputPumpV1DeliveryFailureIsExplicitAndTerminal(t *testing.T) { if result.Status != failedPTYOutputStatusV1(ptyOutputDeliveryFailureReasonV1) || !errors.Is(result.Err, deliveryErr) { t.Fatalf("Finalize() = %#v", result) } + if !IsPTYOutputDeliveryFailureV1(result) || IsPTYOutputObservationFailureV1(result) { + t.Fatalf("delivery failure classification = %#v", result) + } if calls.Load() != 1 { t.Fatalf("emitter calls = %d", calls.Load()) } @@ -157,6 +160,9 @@ func TestPTYOutputPumpV1DeliversBytesReturnedWithReadFailure(t *testing.T) { if result.Status != failedPTYOutputStatusV1(ptyOutputReadFailureReasonV1) || !errors.Is(result.Err, readErr) { t.Fatalf("Finalize() = %#v", result) } + if IsPTYOutputDeliveryFailureV1(result) || !IsPTYOutputObservationFailureV1(result) { + t.Fatalf("read failure classification = %#v", result) + } if source.closeCount.Load() != 1 { t.Fatalf("source close count = %d", source.closeCount.Load()) } @@ -335,6 +341,9 @@ func TestPTYOutputPumpV1DeadlineIncludesEarlierWorkloadShutdown(t *testing.T) { if result.Status != failedPTYOutputStatusV1(ptyOutputTimeoutReasonV1) { t.Fatalf("Finalize() = %#v", result) } + if IsPTYOutputDeliveryFailureV1(result) || IsPTYOutputObservationFailureV1(result) { + t.Fatalf("timeout failure classification = %#v", result) + } _ = writer.Close() } @@ -482,6 +491,9 @@ func TestPTYOutputPumpV1CloseFailureCannotReportDrained(t *testing.T) { if result.Status != failedPTYOutputStatusV1(ptyOutputClosureFailureReasonV1) || !errors.Is(result.Err, closeErr) { t.Fatalf("Finalize() = %#v", result) } + if IsPTYOutputDeliveryFailureV1(result) || !IsPTYOutputObservationFailureV1(result) { + t.Fatalf("closure failure classification = %#v", result) + } } type closeErrorReaderV1 struct { diff --git a/internal/controlledsession/session_io.go b/internal/controlledsession/session_io.go index 28544cc..f37b3fe 100644 --- a/internal/controlledsession/session_io.go +++ b/internal/controlledsession/session_io.go @@ -93,6 +93,14 @@ type SessionIOBridgeV1 struct { requestResultMu sync.Mutex requestResult error + + activeRequestMu sync.Mutex + activeRequest *sessionActiveRequestV1 +} + +type sessionActiveRequestV1 struct { + cancel context.CancelFunc + canceledByHost bool } // StartSessionIOBridgeV1 starts request dispatch and PTY output delivery @@ -158,6 +166,20 @@ func (bridge *SessionIOBridgeV1) StopRequests() { bridge.stopOnce.Do(bridge.cancelRequest) } +// CancelActiveRequest cancels the request currently executing without +// stopping request dispatch. This lets a lifecycle owner interrupt an input +// write blocked by PTY backpressure while continuing to accept the terminal +// complete and acknowledgement requests required during teardown. +func (bridge *SessionIOBridgeV1) CancelActiveRequest() { + bridge.activeRequestMu.Lock() + defer bridge.activeRequestMu.Unlock() + if bridge.activeRequest == nil { + return + } + bridge.activeRequest.canceledByHost = true + bridge.activeRequest.cancel() +} + // 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 { @@ -178,6 +200,10 @@ func (bridge *SessionIOBridgeV1) OutputDone() <-chan struct{} { return bridge.output.Done() } +func (bridge *SessionIOBridgeV1) OutputTerminalResult() (PTYOutputFinalizationV1, bool) { + return bridge.output.TerminalResult() +} + func (bridge *SessionIOBridgeV1) FinalizeOutput(deadline time.Time) (PTYOutputFinalizationV1, error) { return bridge.output.Finalize(deadline) } @@ -190,7 +216,24 @@ func (bridge *SessionIOBridgeV1) runRequests(handle ControllerRequestHandlerV1) bridge.setRequestResult(bridge.requestFailure("read", err)) return } - if err := handle(bridge.requestCtx, request); err != nil { + requestCtx, cancelRequest := context.WithCancel(bridge.requestCtx) + active := &sessionActiveRequestV1{cancel: cancelRequest} + bridge.activeRequestMu.Lock() + bridge.activeRequest = active + bridge.activeRequestMu.Unlock() + + err = handle(requestCtx, request) + cancelRequest() + bridge.activeRequestMu.Lock() + canceledByHost := active.canceledByHost + if bridge.activeRequest == active { + bridge.activeRequest = nil + } + bridge.activeRequestMu.Unlock() + if err != nil && canceledByHost { + continue + } + if err != nil { bridge.setRequestResult(bridge.requestFailure("handle", err)) return } diff --git a/internal/controlledsession/session_io_test.go b/internal/controlledsession/session_io_test.go index e1ebf71..4340682 100644 --- a/internal/controlledsession/session_io_test.go +++ b/internal/controlledsession/session_io_test.go @@ -170,6 +170,46 @@ func TestSessionIOBridgeV1SurfacesDisconnectAndHandlerFailure(t *testing.T) { }) } +func TestSessionIOBridgeV1CancelsActiveRequestAndContinuesDispatch(t *testing.T) { + transport := newBridgeTestTransportV1() + inputStarted := make(chan struct{}) + continued := make(chan RequestV1, 1) + bridge, err := StartSessionIOBridgeV1(transport, io.NopCloser(bytes.NewReader(nil)), func(ctx context.Context, request RequestV1) error { + if request.Kind == RequestInputV1 { + close(inputStarted) + <-ctx.Done() + return errors.New("Docker resize connection closed after cancellation") + } + continued <- request + return nil + }) + if err != nil { + t.Fatal(err) + } + transport.requests <- bridgeTestRequestResultV1{request: RequestV1{Kind: RequestInputV1, Bytes: []byte("blocked")}} + select { + case <-inputStarted: + case <-time.After(time.Second): + t.Fatal("input request did not start") + } + bridge.CancelActiveRequest() + transport.requests <- bridgeTestRequestResultV1{request: RequestV1{Kind: RequestCompleteV1}} + select { + case request := <-continued: + if request.Kind != RequestCompleteV1 { + t.Fatalf("continued request = %#v", request) + } + case <-time.After(time.Second): + t.Fatal("request dispatch did not continue after canceling active input") + } + bridge.StopRequests() + waitCtx, cancel := context.WithTimeout(context.Background(), time.Second) + defer cancel() + if err := bridge.WaitRequests(waitCtx); err != nil { + t.Fatalf("WaitRequests() = %v", err) + } +} + func TestSessionIOBridgeV1LifecycleRejectionPreventsPTYEffect(t *testing.T) { authorization := testAuthorizationV1() authorization.Operations = []OperationV1{OperationResizeV1, OperationTerminateV1} diff --git a/internal/dockerdeploy/controlled_session_controller.go b/internal/dockerdeploy/controlled_session_controller.go index 1854ea1..1d6fb17 100644 --- a/internal/dockerdeploy/controlled_session_controller.go +++ b/internal/dockerdeploy/controlled_session_controller.go @@ -89,7 +89,7 @@ func prepareDockerControllerV1( } return nil, errors.Join(createErr, cleanupErr) } - containerID, err := parseDockerControllerContainerIDV1(createOutput.String()) + containerID, err := parseDockerContainerIDV1(createOutput.String()) if err != nil { return nil, fmt.Errorf("create controlled-session controller container %q: %w; refusing name-based cleanup because the created container identity is unknown", plan.Container, err) } @@ -261,7 +261,7 @@ func rollbackControlledSessionControllerContainerV1( return backend.run(cleanup, RunOptions{Context: cleanupCtx}) } -func parseDockerControllerContainerIDV1(output string) (string, error) { +func parseDockerContainerIDV1(output string) (string, error) { containerID := string(bytes.TrimSpace([]byte(output))) if len(containerID) != 64 { return "", fmt.Errorf("Docker create returned invalid full container ID %q", containerID) diff --git a/internal/dockerdeploy/controlled_session_controller_integration_test.go b/internal/dockerdeploy/controlled_session_controller_integration_test.go index 2358495..408e7fd 100644 --- a/internal/dockerdeploy/controlled_session_controller_integration_test.go +++ b/internal/dockerdeploy/controlled_session_controller_integration_test.go @@ -242,8 +242,10 @@ func controlledSessionControllerIntegrationPlanV1( ContainerDirectory: controlledSessionChannelRootV1, ContainerSocket: path.Join(controlledSessionChannelRootV1, controlledSessionChannelSocketNameV1), } - uid := os.Geteuid() - gid := os.Getegid() + _, uid, gid, _, err := currentHostRuntimeIdentityV1() + if err != nil { + t.Fatal(err) + } identity := RuntimeUserPlan{LocalUser: "reploy", UID: uid, GID: gid, DockerUser: fmt.Sprintf("%d:%d", uid, gid)} controllerCurrent := CurrentBuild{Generation: deploy.EnvironmentGenerationState{ Reference: image, BuildLockDigest: canonical.Digest("sha256:" + strings.Repeat("4", 64)), diff --git a/internal/dockerdeploy/controlled_session_controller_test.go b/internal/dockerdeploy/controlled_session_controller_test.go index 50f6a8b..fb96f3b 100644 --- a/internal/dockerdeploy/controlled_session_controller_test.go +++ b/internal/dockerdeploy/controlled_session_controller_test.go @@ -174,11 +174,11 @@ func TestPrepareDockerControllerV1RollsBackOwnedAmbiguousCreateFailure(t *testin } func TestParseDockerControllerContainerIDV1(t *testing.T) { - if got, err := parseDockerControllerContainerIDV1("\n" + dockerControllerTestContainerIDV1 + "\n"); err != nil || got != dockerControllerTestContainerIDV1 { + if got, err := parseDockerContainerIDV1("\n" + dockerControllerTestContainerIDV1 + "\n"); err != nil || got != dockerControllerTestContainerIDV1 { t.Fatalf("valid container ID = %q, %v", got, err) } for _, invalid := range []string{"", "abc", strings.Repeat("g", 64)} { - if _, err := parseDockerControllerContainerIDV1(invalid); err == nil { + if _, err := parseDockerContainerIDV1(invalid); err == nil { t.Fatalf("invalid container ID %q was accepted", invalid) } } diff --git a/internal/dockerdeploy/controlled_session_supervisor.go b/internal/dockerdeploy/controlled_session_supervisor.go new file mode 100644 index 0000000..1daa4c7 --- /dev/null +++ b/internal/dockerdeploy/controlled_session_supervisor.go @@ -0,0 +1,870 @@ +package dockerdeploy + +import ( + "context" + "errors" + "fmt" + "io" + "sync" + "time" + + "github.com/omry/reploy/internal/controlledsession" +) + +const controlledSessionOutputFinalizationTimeoutV1 = time.Duration(controlledsession.DefaultOutputFinalizationTimeoutMillisecondsV1) * time.Millisecond + +type ControlledSessionRunOptionsV1 struct { + StartupTimeout time.Duration + TerminationGrace time.Duration + ControllerFinalizationTimeout time.Duration + ResultAcknowledgementTimeout time.Duration + CleanupTimeout time.Duration +} + +// ControlledSessionRunResultV1 separates the authoritative session result +// delivered over the private channel from delivery-tail facts that can exist +// only after that channel and the controller have been removed. +type ControlledSessionRunResultV1 struct { + SessionResult controlledsession.ResultV1 + ResultDelivered bool + ResultAcknowledged bool + ControllerStatus controlledsession.ProcessStatusV1 + DeliveryTailCleanupStatus controlledsession.CleanupStatusV1 + DeliveryTailRecoveryAction controlledsession.RecoveryActionV1 +} + +type controlledSessionControllerRuntimeV1 interface { + Start(context.Context) error + Wait(context.Context) (controlledsession.ProcessStatusV1, error) + RequestGracefulStop(context.Context) error + ForceStop(context.Context) error + Cleanup(context.Context) error +} + +type controlledSessionWorkloadRuntimeV1 interface { + controlledsession.WorkloadPTYControlV1 + Output() (io.ReadCloser, error) + Start(context.Context) error + Started() bool + Wait(context.Context) (controlledsession.ProcessStatusV1, error) + RequestGracefulStop(context.Context) error + ForceStop(context.Context) error + Close() error + Cleanup(context.Context) error +} + +type controlledSessionChannelRuntimeV1 interface { + Claim(context.Context) (controlledsession.ControllerTransportV1, error) + Close() error +} + +type privateControlledSessionChannelRuntimeV1 struct { + channel *controlledsession.PrivateChannelV1 +} + +func (runtime *privateControlledSessionChannelRuntimeV1) Claim(ctx context.Context) (controlledsession.ControllerTransportV1, error) { + return runtime.channel.Claim(ctx) +} + +func (runtime *privateControlledSessionChannelRuntimeV1) Close() error { + return runtime.channel.Close() +} + +type controlledSessionSupervisorBackendV1 struct { + prepareChannel func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) + prepareController func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) + prepareWorkload func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) + now func() time.Time +} + +type controlledSessionProcessResultV1 struct { + status controlledsession.ProcessStatusV1 + err error +} + +type controlledSessionSupervisorV1 struct { + plan ControlledSessionExecutionPlanV1 + options ControlledSessionRunOptionsV1 + backend controlledSessionSupervisorBackendV1 + machine *controlledsession.MachineV1 + + channel controlledSessionChannelRuntimeV1 + controller controlledSessionControllerRuntimeV1 + workload controlledSessionWorkloadRuntimeV1 + bridge *controlledsession.SessionIOBridgeV1 + + startupResolved chan struct{} + resolveOnce sync.Once + stateChanged chan struct{} + resultDeliveryStarted chan struct{} + resultDeliveryResolved chan struct{} + startDeliveryOnce sync.Once + resolveDeliveryOnce sync.Once + outputPublicationStarted chan struct{} + outputPublicationResolved chan struct{} + startOutputPublicationOnce sync.Once + resolveOutputPublicationOnce sync.Once + outputPublicationSucceeded bool + terminationMu sync.Mutex + terminationAt time.Time + + workloadResult <-chan controlledSessionProcessResultV1 + controllerResult <-chan controlledSessionProcessResultV1 + workloadObserved *controlledSessionProcessResultV1 + controllerObserved *controlledSessionProcessResultV1 + workloadRecorded bool + controllerStarted bool + workloadStarted bool + + transportHealthy bool + diagnosticErr error +} + +// RunControlledSessionV1 owns one attached controller/workload operation from +// inert resource creation through terminal acknowledgement and ordinary +// delivery-tail cleanup. Crash reconciliation, watchdog ownership, networking, +// and public command exposure are deliberately outside this lifecycle core. +func RunControlledSessionV1( + ctx context.Context, + plan ControlledSessionExecutionPlanV1, + options ControlledSessionRunOptionsV1, +) (ControlledSessionRunResultV1, error) { + return runControlledSessionV1(ctx, plan, options, controlledSessionSupervisorBackendV1{ + prepareChannel: func(plan ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { + channel, err := PrepareControlledSessionChannelV1(plan) + if err != nil { + return nil, err + } + return &privateControlledSessionChannelRuntimeV1{channel: channel}, nil + }, + prepareController: func(ctx context.Context, plan ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { + return PrepareDockerControllerV1(ctx, plan) + }, + prepareWorkload: func(ctx context.Context, plan ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { + return PrepareDockerWorkloadPTYV1(ctx, plan) + }, + now: time.Now, + }) +} + +func runControlledSessionV1( + ctx context.Context, + plan ControlledSessionExecutionPlanV1, + options ControlledSessionRunOptionsV1, + backend controlledSessionSupervisorBackendV1, +) (ControlledSessionRunResultV1, error) { + if ctx == nil || ctx.Done() == nil { + return ControlledSessionRunResultV1{}, fmt.Errorf("run controlled session: cancelable host context is required") + } + if err := ValidateControlledSessionExecutionPlanV1(plan); err != nil { + return ControlledSessionRunResultV1{}, fmt.Errorf("run controlled session plan: %w", err) + } + if err := validateControlledSessionRunOptionsV1(options); err != nil { + return ControlledSessionRunResultV1{}, err + } + if backend.prepareChannel == nil || backend.prepareController == nil || backend.prepareWorkload == nil || backend.now == nil { + return ControlledSessionRunResultV1{}, fmt.Errorf("run controlled session: supervisor backend is incomplete") + } + machine, err := controlledsession.NewMachineV1(plan.Authorization) + if err != nil { + return ControlledSessionRunResultV1{}, fmt.Errorf("run controlled session lifecycle: %w", err) + } + supervisor := &controlledSessionSupervisorV1{ + plan: plan, options: options, backend: backend, machine: machine, + startupResolved: make(chan struct{}), stateChanged: make(chan struct{}, 1), + resultDeliveryStarted: make(chan struct{}), resultDeliveryResolved: make(chan struct{}), + outputPublicationStarted: make(chan struct{}), outputPublicationResolved: make(chan struct{}), + transportHealthy: true, + } + return supervisor.run(ctx) +} + +func validateControlledSessionRunOptionsV1(options ControlledSessionRunOptionsV1) error { + for _, value := range []struct { + name string + value time.Duration + }{ + {name: "startup timeout", value: options.StartupTimeout}, + {name: "termination grace", value: options.TerminationGrace}, + {name: "controller finalization timeout", value: options.ControllerFinalizationTimeout}, + {name: "result acknowledgement timeout", value: options.ResultAcknowledgementTimeout}, + {name: "cleanup timeout", value: options.CleanupTimeout}, + } { + if value.value <= 0 { + return fmt.Errorf("run controlled session: %s must be finite and positive", value.name) + } + } + return nil +} + +func (supervisor *controlledSessionSupervisorV1) run(ctx context.Context) (ControlledSessionRunResultV1, error) { + startupCtx, cancelStartup := context.WithTimeout(ctx, supervisor.options.StartupTimeout) + defer cancelStartup() + if err := supervisor.prepare(startupCtx); err != nil { + if ctx.Err() != nil { + _, observeErr := supervisor.observe(controlledsession.ObservationV1{ + Kind: controlledsession.ObservationHostCancelV1, Reason: "host operation was canceled during startup", + }) + if observeErr != nil { + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, fmt.Errorf("record controlled-session startup cancellation: %w", observeErr)) + } + } + return supervisor.finishStartupFailure(err) + } + + if _, err := supervisor.observe(controlledsession.ObservationV1{Kind: controlledsession.ObservationActivatedV1}); err != nil { + return supervisor.finishStartupFailure(fmt.Errorf("activate controlled-session lifecycle: %w", err)) + } + supervisor.resolveStartup() + + supervisor.waitForTermination(ctx) + supervisor.bridge.CancelActiveRequest() + supervisor.sendTerminating() + supervisor.stopAndObserveWorkload() + supervisor.finalizeWorkloadOutput() + supervisor.waitForControllerFinalization() + + preDeliveryCleanup, preDeliveryRecovery, cleanupErr := supervisor.cleanupWorkload() + finish := supervisor.finishStatus(preDeliveryCleanup, preDeliveryRecovery) + transition, finishErr := supervisor.observe(controlledsession.ObservationV1{Kind: controlledsession.ObservationFinishedV1, Finish: &finish}) + if finishErr != nil { + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, fmt.Errorf("finish controlled-session lifecycle: %w", finishErr)) + } + result := transition.Result + if result == nil { + result = supervisor.machine.Snapshot().Result + } + invocation := supervisor.deliverTerminalResult(result) + + invocation.ControllerStatus, invocation.DeliveryTailCleanupStatus, + invocation.DeliveryTailRecoveryAction = supervisor.cleanupDeliveryTail() + return invocation, errors.Join(supervisor.diagnosticErr, cleanupErr) +} + +func (supervisor *controlledSessionSupervisorV1) prepare(ctx context.Context) error { + channel, err := supervisor.backend.prepareChannel(supervisor.plan) + if err != nil { + return fmt.Errorf("prepare controlled-session channel: %w", err) + } + supervisor.channel = channel + + controller, err := supervisor.backend.prepareController(ctx, supervisor.plan.Controller) + if err != nil { + return fmt.Errorf("prepare controlled-session controller: %w", err) + } + supervisor.controller = controller + workload, err := supervisor.backend.prepareWorkload(ctx, supervisor.plan.Workload) + if err != nil { + return fmt.Errorf("prepare controlled-session workload: %w", err) + } + supervisor.workload = workload + output, err := workload.Output() + if err != nil { + return fmt.Errorf("claim controlled-session workload output: %w", err) + } + if err := controller.Start(ctx); err != nil { + return fmt.Errorf("start controlled-session controller: %w", err) + } + supervisor.controllerStarted = true + supervisor.controllerResult = observeControlledSessionProcessV1(controller.Wait) + transport, err := channel.Claim(ctx) + if err != nil { + return fmt.Errorf("claim controlled-session controller channel: %w", err) + } + bridge, err := controlledsession.StartSessionIOBridgeV1(transport, output, supervisor.handleRequest) + if err != nil { + return fmt.Errorf("start controlled-session I/O bridge: %w", err) + } + supervisor.bridge = bridge + if err := workload.Start(ctx); err != nil { + supervisor.workloadStarted = workload.Started() + if supervisor.workloadStarted { + supervisor.workloadResult = observeControlledSessionProcessV1(workload.Wait) + } + return fmt.Errorf("start controlled-session workload: %w", err) + } + supervisor.workloadStarted = true + supervisor.workloadResult = observeControlledSessionProcessV1(workload.Wait) + return nil +} + +func observeControlledSessionProcessV1( + wait func(context.Context) (controlledsession.ProcessStatusV1, error), +) <-chan controlledSessionProcessResultV1 { + result := make(chan controlledSessionProcessResultV1, 1) + go func() { + status, err := wait(context.Background()) + result <- controlledSessionProcessResultV1{status: status, err: err} + }() + return result +} + +func (supervisor *controlledSessionSupervisorV1) handleRequest(ctx context.Context, request controlledsession.RequestV1) error { + select { + case <-supervisor.startupResolved: + case <-ctx.Done(): + return ctx.Err() + } + if request.Kind == controlledsession.RequestAcknowledgeTerminatedV1 { + select { + case <-supervisor.resultDeliveryStarted: + select { + case <-supervisor.resultDeliveryResolved: + case <-ctx.Done(): + return ctx.Err() + } + default: + } + } + if request.Kind == controlledsession.RequestCompleteV1 { + select { + case <-supervisor.outputPublicationStarted: + select { + case <-supervisor.outputPublicationResolved: + case <-ctx.Done(): + return ctx.Err() + } + if !supervisor.outputPublicationSucceeded { + return fmt.Errorf("controlled-session workload output finalization event was not published") + } + default: + } + } + transition, err := supervisor.machine.ApplyRequest(request) + if err != nil { + return err + } + supervisor.recordTransition(transition) + if _, err := controlledsession.ApplyAcceptedWorkloadPTYRequestV1(ctx, supervisor.workload, request); err != nil { + return err + } + supervisor.notifyStateChanged() + return nil +} + +func (supervisor *controlledSessionSupervisorV1) waitForTermination(ctx context.Context) { + outputDone := supervisor.bridge.OutputDone() + for supervisor.machine.Snapshot().State == controlledsession.StateActiveV1 { + select { + case result := <-supervisor.workloadResult: + supervisor.workloadObserved = &result + supervisor.observeWorkloadResult(result) + case result := <-supervisor.controllerResult: + supervisor.controllerObserved = &result + supervisor.observeControllerLoss("controller process exited", result.err) + case <-supervisor.bridge.RequestsDone(): + supervisor.observeRequestFailure() + case <-outputDone: + outputDone = nil + supervisor.observeOutputTermination() + case <-supervisor.stateChanged: + case <-ctx.Done(): + _, err := supervisor.observe(controlledsession.ObservationV1{ + Kind: controlledsession.ObservationHostCancelV1, Reason: "host operation was canceled", + }) + if err != nil { + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, fmt.Errorf("record controlled-session host cancellation: %w", err)) + } + } + } +} + +func (supervisor *controlledSessionSupervisorV1) observeOutputTermination() { + result, stopped := supervisor.bridge.OutputTerminalResult() + if !stopped { + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, fmt.Errorf("controlled-session output completion was signaled without a terminal result")) + return + } + supervisor.observeOutputFailure(result) +} + +func (supervisor *controlledSessionSupervisorV1) observeOutputFailure(result controlledsession.PTYOutputFinalizationV1) { + if result.Status.Kind != controlledsession.WorkloadOutputFinalizationFailedV1 { + return + } + if controlledsession.IsPTYOutputDeliveryFailureV1(result) { + if supervisor.machine.Snapshot().ControllerFinalizationStatus.Kind == controlledsession.ControllerFinalizationLostV1 { + return + } + supervisor.transportHealthy = false + supervisor.observeControllerLoss("controller event transport was lost", nil) + return + } + if !controlledsession.IsPTYOutputObservationFailureV1(result) || + supervisor.machine.Snapshot().RuntimeObservationStatus.Kind == controlledsession.RuntimeObservationLostV1 { + return + } + _, err := supervisor.observe(controlledsession.ObservationV1{ + Kind: controlledsession.ObservationRuntimeObservationLostV1, + Reason: "workload output observation was lost", + }) + if err != nil { + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, fmt.Errorf("record controlled-session output observation loss: %w", err)) + } +} + +func (supervisor *controlledSessionSupervisorV1) observeWorkloadResult(result controlledSessionProcessResultV1) { + if supervisor.workloadRecorded { + return + } + supervisor.workloadRecorded = true + if result.err != nil { + _, err := supervisor.observe(controlledsession.ObservationV1{ + Kind: controlledsession.ObservationRuntimeObservationLostV1, Reason: "workload runtime observation was lost", + }) + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, result.err, err) + return + } + _, err := supervisor.observe(controlledsession.ObservationV1{ + Kind: controlledsession.ObservationWorkloadExitV1, WorkloadStatus: &result.status, + }) + if err != nil { + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, fmt.Errorf("record controlled-session workload exit: %w", err)) + return + } + if supervisor.transportHealthy { + if err := supervisor.sendPreFinalizationLifecycleEvent(controlledsession.EventV1{ + Kind: controlledsession.EventWorkloadExitV1, + WorkloadExit: &controlledsession.WorkloadExitV1{Status: result.status}, + }); err != nil { + supervisor.loseTransport("send workload exit", err) + } + } +} + +func (supervisor *controlledSessionSupervisorV1) observeControllerLoss(reason string, detail error) { + snapshot := supervisor.machine.Snapshot() + if snapshot.State == controlledsession.StateTerminatedV1 || snapshot.ControllerFinalizationStatus.Kind == controlledsession.ControllerFinalizationLostV1 { + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, detail) + return + } + _, err := supervisor.observe(controlledsession.ObservationV1{ + Kind: controlledsession.ObservationControllerLostV1, Reason: reason, + }) + supervisor.transportHealthy = false + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, detail, err) +} + +func (supervisor *controlledSessionSupervisorV1) observeRequestFailure() { + waitCtx, cancel := context.WithTimeout(context.Background(), supervisor.options.CleanupTimeout) + defer cancel() + err := supervisor.bridge.WaitRequests(waitCtx) + if err == nil { + err = fmt.Errorf("controller request stream stopped before host teardown") + } + supervisor.observeControllerLoss("controller request stream was lost", err) +} + +func (supervisor *controlledSessionSupervisorV1) sendTerminating() { + if !supervisor.transportHealthy || supervisor.bridge == nil { + return + } + cause := supervisor.machine.Snapshot().Cause + if err := supervisor.sendPreFinalizationLifecycleEvent(controlledsession.EventV1{ + Kind: controlledsession.EventTerminatingV1, Terminating: &controlledsession.TerminatingV1{Cause: cause}, + }); err != nil { + supervisor.loseTransport("send terminating event", err) + } +} + +func (supervisor *controlledSessionSupervisorV1) stopAndObserveWorkload() { + if !supervisor.workloadStarted { + return + } + if supervisor.workloadObserved != nil && supervisor.workloadObserved.err == nil { + supervisor.observeWorkloadResult(*supervisor.workloadObserved) + return + } + + finalizationDeadline := supervisor.terminationDeadline() + graceDeadline := earlierControlledSessionDeadlineV1( + time.Now().Add(supervisor.options.TerminationGrace), + finalizationDeadline, + ) + stopCtx, cancel := context.WithDeadline(context.Background(), graceDeadline) + stopErr := supervisor.workload.RequestGracefulStop(stopCtx) + cancel() + if supervisor.workloadObserved == nil { + timer := time.NewTimer(time.Until(graceDeadline)) + select { + case result := <-supervisor.workloadResult: + timer.Stop() + supervisor.workloadObserved = &result + if result.err == nil { + supervisor.observeWorkloadResult(result) + return + } + supervisor.observeWorkloadResult(result) + case <-timer.C: + } + } + + forceCtx, forceCancel := context.WithDeadline(context.Background(), finalizationDeadline) + forceErr := supervisor.workload.ForceStop(forceCtx) + forceCancel() + if supervisor.workloadObserved == nil { + waitTimer := time.NewTimer(time.Until(finalizationDeadline)) + select { + case result := <-supervisor.workloadResult: + supervisor.workloadObserved = &result + waitTimer.Stop() + if result.err == nil { + supervisor.observeWorkloadResult(result) + return + } + supervisor.observeWorkloadResult(result) + case <-waitTimer.C: + result := controlledSessionProcessResultV1{ + status: controlledsession.ProcessStatusV1{Kind: controlledsession.ProcessStatusUnavailableV1, Reason: "workload runtime observation was lost"}, + err: fmt.Errorf("timed out observing controlled-session workload after forced stop"), + } + supervisor.workloadObserved = &result + supervisor.observeWorkloadResult(result) + } + } + if supervisor.workloadObserved == nil || supervisor.workloadObserved.err != nil { + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, stopErr, forceErr) + } +} + +func (supervisor *controlledSessionSupervisorV1) finalizeWorkloadOutput() { + deadline := supervisor.terminationDeadline() + finalization, err := supervisor.bridge.FinalizeOutput(deadline) + if err != nil { + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, err) + return + } + supervisor.observeOutputFailure(finalization) + if supervisor.machine.Snapshot().RuntimeObservationStatus.Kind == controlledsession.RuntimeObservationLostV1 && + finalization.Status.Kind == controlledsession.WorkloadOutputFinalizationDrainedV1 { + finalization.Status = controlledsession.WorkloadOutputFinalizationStatusV1{ + Kind: controlledsession.WorkloadOutputFinalizationFailedV1, + Reason: "workload runtime observation was lost before output finalization completed", + } + } + supervisor.startOutputPublicationOnce.Do(func() { close(supervisor.outputPublicationStarted) }) + defer supervisor.resolveOutputPublication() + _, observeErr := supervisor.observe(controlledsession.ObservationV1{ + Kind: controlledsession.ObservationWorkloadOutputsFinalizedV1, + WorkloadOutputFinalizationStatus: &finalization.Status, + }) + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, finalization.Err, observeErr) + if observeErr != nil { + return + } + if !supervisor.transportHealthy { + return + } + if err := supervisor.sendLifecycleEvent(controlledsession.EventV1{ + Kind: controlledsession.EventWorkloadOutputsFinalizedV1, + WorkloadOutputsFinalized: &controlledsession.WorkloadOutputsFinalizedV1{ + Status: finalization.Status.Kind, Reason: finalization.Status.Reason, + }, + }); err != nil { + supervisor.loseTransport("send workload output finalization", err) + return + } + if supervisor.machine.Snapshot().AwaitingWorkloadOutputPublication { + if _, err := supervisor.observe(controlledsession.ObservationV1{Kind: controlledsession.ObservationWorkloadOutputsPublishedV1}); err != nil { + supervisor.observeControllerLoss("workload output finalization publication could not be recorded", err) + return + } + } + supervisor.outputPublicationSucceeded = true + supervisor.resolveOutputPublication() +} + +func (supervisor *controlledSessionSupervisorV1) resolveOutputPublication() { + supervisor.resolveOutputPublicationOnce.Do(func() { close(supervisor.outputPublicationResolved) }) +} + +func (supervisor *controlledSessionSupervisorV1) waitForControllerFinalization() { + if !supervisor.machine.Snapshot().AwaitingControllerFinalization { + return + } + timer := time.NewTimer(supervisor.options.ControllerFinalizationTimeout) + defer timer.Stop() + for supervisor.machine.Snapshot().AwaitingControllerFinalization { + select { + case <-supervisor.stateChanged: + case result := <-supervisor.controllerResult: + supervisor.controllerObserved = &result + supervisor.observeControllerLoss("controller process exited during finalization", result.err) + case <-supervisor.bridge.RequestsDone(): + supervisor.observeRequestFailure() + case <-timer.C: + _, err := supervisor.observe(controlledsession.ObservationV1{Kind: controlledsession.ObservationControllerFinalizationExpiredV1}) + if err != nil && supervisor.machine.Snapshot().AwaitingControllerFinalization { + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, fmt.Errorf("expire controlled-session controller finalization: %w", err)) + } + return + } + } +} + +func (supervisor *controlledSessionSupervisorV1) cleanupWorkload() ( + controlledsession.CleanupStatusV1, + controlledsession.RecoveryActionV1, + error, +) { + closeErr := supervisor.workload.Close() + cleanupCtx, cancel := context.WithTimeout(context.Background(), supervisor.options.CleanupTimeout) + cleanupErr := supervisor.workload.Cleanup(cleanupCtx) + cancel() + err := errors.Join(closeErr, cleanupErr) + if err == nil { + return controlledsession.CleanupStatusV1{Kind: controlledsession.CleanupStatusSucceededV1}, controlledsession.RecoveryNoneV1, nil + } + return controlledsession.CleanupStatusV1{ + Kind: controlledsession.CleanupStatusFailedV1, Message: "controlled-session workload cleanup failed", + }, controlledsession.RecoveryRetryCleanupV1, err +} + +func (supervisor *controlledSessionSupervisorV1) finishStatus( + cleanup controlledsession.CleanupStatusV1, + recovery controlledsession.RecoveryActionV1, +) controlledsession.FinishV1 { + snapshot := supervisor.machine.Snapshot() + controller := snapshot.ControllerFinalizationStatus + if controller.Kind == controlledsession.ControllerFinalizationActiveV1 || controller.Kind == controlledsession.ControllerFinalizationUnknownV1 { + controller = controlledsession.ControllerFinalizationStatusV1{Kind: controlledsession.ControllerFinalizationNotCompletedV1} + } + workload := snapshot.WorkloadStatus + if workload.Kind == controlledsession.ProcessStatusUnknownV1 && snapshot.RuntimeObservationStatus.Kind == controlledsession.RuntimeObservationLostV1 { + workload = controlledsession.ProcessStatusV1{Kind: controlledsession.ProcessStatusUnavailableV1, Reason: "workload runtime observation was lost"} + } + return controlledsession.FinishV1{ + WorkloadStatus: workload, + WorkloadOutputFinalizationStatus: snapshot.WorkloadOutputFinalizationStatus, + ControllerFinalizationStatus: controller, + CleanupStatus: cleanup, + RecoveryAction: recovery, + } +} + +func (supervisor *controlledSessionSupervisorV1) waitForResultAcknowledgement() bool { + timer := time.NewTimer(supervisor.options.ResultAcknowledgementTimeout) + defer timer.Stop() + for supervisor.machine.Snapshot().AwaitingResultAcknowledgement { + select { + case <-supervisor.stateChanged: + case <-supervisor.bridge.RequestsDone(): + return supervisor.machine.Snapshot().ResultAcknowledged + case result := <-supervisor.controllerResult: + supervisor.controllerObserved = &result + case <-timer.C: + return supervisor.machine.Snapshot().ResultAcknowledged + } + } + return supervisor.machine.Snapshot().ResultAcknowledged +} + +func (supervisor *controlledSessionSupervisorV1) cleanupDeliveryTail() ( + controlledsession.ProcessStatusV1, + controlledsession.CleanupStatusV1, + controlledsession.RecoveryActionV1, +) { + if supervisor.bridge != nil { + supervisor.bridge.StopRequests() + } + var cleanupErr error + if supervisor.channel != nil { + cleanupErr = errors.Join(cleanupErr, supervisor.channel.Close()) + } + if supervisor.controller != nil { + if supervisor.controllerStarted && supervisor.controllerObserved == nil { + select { + case result := <-supervisor.controllerResult: + supervisor.controllerObserved = &result + default: + } + } + if supervisor.controllerStarted && supervisor.controllerObserved == nil { + graceDeadline := time.Now().Add(supervisor.options.TerminationGrace) + stopCtx, cancel := context.WithDeadline(context.Background(), graceDeadline) + stopErr := supervisor.controller.RequestGracefulStop(stopCtx) + cancel() + timer := time.NewTimer(time.Until(graceDeadline)) + select { + case result := <-supervisor.controllerResult: + supervisor.controllerObserved = &result + timer.Stop() + if result.err != nil { + cleanupErr = errors.Join(cleanupErr, stopErr) + } + case <-timer.C: + forceCtx, forceCancel := context.WithTimeout(context.Background(), supervisor.options.CleanupTimeout) + forceErr := supervisor.controller.ForceStop(forceCtx) + forceCancel() + waitTimer := time.NewTimer(supervisor.options.CleanupTimeout) + select { + case result := <-supervisor.controllerResult: + supervisor.controllerObserved = &result + waitTimer.Stop() + if result.err != nil { + cleanupErr = errors.Join(cleanupErr, stopErr, forceErr) + } + case <-waitTimer.C: + cleanupErr = errors.Join(cleanupErr, stopErr, forceErr, fmt.Errorf("timed out observing controlled-session controller after forced stop")) + } + } + } + cleanupCtx, cleanupCancel := context.WithTimeout(context.Background(), supervisor.options.CleanupTimeout) + cleanupErr = errors.Join(cleanupErr, supervisor.controller.Cleanup(cleanupCtx)) + cleanupCancel() + } + status := controlledsession.ProcessStatusV1{Kind: controlledsession.ProcessStatusUnknownV1} + if supervisor.controllerObserved != nil { + status = supervisor.controllerObserved.status + cleanupErr = errors.Join(cleanupErr, supervisor.controllerObserved.err) + } + if cleanupErr != nil { + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, cleanupErr) + return status, controlledsession.CleanupStatusV1{ + Kind: controlledsession.CleanupStatusFailedV1, Message: "controlled-session delivery-tail cleanup failed", + }, controlledsession.RecoveryRetryCleanupV1 + } + return status, controlledsession.CleanupStatusV1{Kind: controlledsession.CleanupStatusSucceededV1}, controlledsession.RecoveryNoneV1 +} + +func (supervisor *controlledSessionSupervisorV1) finishStartupFailure(cause error) (ControlledSessionRunResultV1, error) { + var finalizedOutput *controlledsession.WorkloadOutputFinalizationStatusV1 + if !supervisor.workloadStarted && supervisor.bridge != nil && supervisor.workload != nil { + closeErr := supervisor.workload.Close() + finalization, finalizationErr := supervisor.bridge.FinalizeOutput( + time.Now().Add(supervisor.options.CleanupTimeout), + ) + finalizedOutput = &finalization.Status + supervisor.diagnosticErr = errors.Join( + supervisor.diagnosticErr, + closeErr, + finalization.Err, + finalizationErr, + ) + } + if supervisor.machine.Snapshot().ControllerFinalizationStatus.Kind == controlledsession.ControllerFinalizationUnknownV1 { + _, observeErr := supervisor.observe(controlledsession.ObservationV1{ + Kind: controlledsession.ObservationStartupFailureV1, + Reason: "controlled-session startup failed", + WorkloadOutputPending: supervisor.workloadStarted, + WorkloadOutputFinalizationStatus: finalizedOutput, + }) + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, observeErr) + } + supervisor.resolveStartup() + if supervisor.workloadStarted { + supervisor.sendTerminating() + supervisor.stopAndObserveWorkload() + supervisor.finalizeWorkloadOutput() + } + preCleanup, recovery, cleanupErr := controlledsession.CleanupStatusV1{Kind: controlledsession.CleanupStatusSucceededV1}, controlledsession.RecoveryNoneV1, error(nil) + if supervisor.workload != nil { + preCleanup, recovery, cleanupErr = supervisor.cleanupWorkload() + } + finish := supervisor.finishStatus(preCleanup, recovery) + transition, finishErr := supervisor.observe(controlledsession.ObservationV1{Kind: controlledsession.ObservationFinishedV1, Finish: &finish}) + invocation := supervisor.deliverTerminalResult(transition.Result) + invocation.ControllerStatus, invocation.DeliveryTailCleanupStatus, + invocation.DeliveryTailRecoveryAction = supervisor.cleanupDeliveryTail() + return invocation, errors.Join(cause, supervisor.diagnosticErr, cleanupErr, finishErr) +} + +func (supervisor *controlledSessionSupervisorV1) deliverTerminalResult(result *controlledsession.ResultV1) ControlledSessionRunResultV1 { + invocation := ControlledSessionRunResultV1{} + if result == nil { + return invocation + } + invocation.SessionResult = *result + if !supervisor.transportHealthy || supervisor.bridge == nil { + return invocation + } + supervisor.startDeliveryOnce.Do(func() { close(supervisor.resultDeliveryStarted) }) + defer supervisor.resolveDelivery() + if err := supervisor.sendLifecycleEvent(controlledsession.EventV1{ + Kind: controlledsession.EventTerminatedV1, Terminated: result, + }); err != nil { + supervisor.loseTransport("deliver terminal result", err) + return invocation + } + invocation.ResultDelivered = true + if _, err := supervisor.observe(controlledsession.ObservationV1{Kind: controlledsession.ObservationResultDeliveredV1}); err != nil { + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, fmt.Errorf("record controlled-session result delivery: %w", err)) + return invocation + } + supervisor.resolveDelivery() + invocation.ResultAcknowledged = supervisor.waitForResultAcknowledgement() + return invocation +} + +func (supervisor *controlledSessionSupervisorV1) resolveDelivery() { + supervisor.resolveDeliveryOnce.Do(func() { close(supervisor.resultDeliveryResolved) }) +} + +func (supervisor *controlledSessionSupervisorV1) observe(observation controlledsession.ObservationV1) (controlledsession.TransitionV1, error) { + transition, err := supervisor.machine.Observe(observation) + if err == nil { + supervisor.recordTransition(transition) + supervisor.notifyStateChanged() + } + return transition, err +} + +func (supervisor *controlledSessionSupervisorV1) recordTransition(transition controlledsession.TransitionV1) { + if !transition.BeginTermination { + return + } + supervisor.terminationMu.Lock() + if supervisor.terminationAt.IsZero() { + supervisor.terminationAt = supervisor.backend.now() + } + supervisor.terminationMu.Unlock() +} + +func (supervisor *controlledSessionSupervisorV1) notifyStateChanged() { + select { + case supervisor.stateChanged <- struct{}{}: + default: + } +} + +func (supervisor *controlledSessionSupervisorV1) resolveStartup() { + supervisor.resolveOnce.Do(func() { close(supervisor.startupResolved) }) +} + +func (supervisor *controlledSessionSupervisorV1) terminationDeadline() time.Time { + supervisor.terminationMu.Lock() + defer supervisor.terminationMu.Unlock() + if supervisor.terminationAt.IsZero() { + supervisor.terminationAt = supervisor.backend.now() + } + return supervisor.terminationAt.Add(controlledSessionOutputFinalizationTimeoutV1) +} + +func (supervisor *controlledSessionSupervisorV1) sendLifecycleEvent(event controlledsession.EventV1) error { + ctx, cancel := context.WithTimeout(context.Background(), supervisor.options.CleanupTimeout) + defer cancel() + return supervisor.bridge.SendLifecycleEvent(ctx, event) +} + +func (supervisor *controlledSessionSupervisorV1) sendPreFinalizationLifecycleEvent(event controlledsession.EventV1) error { + deadline := earlierControlledSessionDeadlineV1( + time.Now().Add(supervisor.options.CleanupTimeout), + supervisor.terminationDeadline(), + ) + ctx, cancel := context.WithDeadline(context.Background(), deadline) + defer cancel() + return supervisor.bridge.SendLifecycleEvent(ctx, event) +} + +func earlierControlledSessionDeadlineV1(left time.Time, right time.Time) time.Time { + if left.Before(right) { + return left + } + return right +} + +func (supervisor *controlledSessionSupervisorV1) loseTransport(action string, err error) { + supervisor.transportHealthy = false + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, fmt.Errorf("%s: %w", action, err)) + supervisor.observeControllerLoss("controller event transport was lost", nil) +} diff --git a/internal/dockerdeploy/controlled_session_supervisor_integration_test.go b/internal/dockerdeploy/controlled_session_supervisor_integration_test.go new file mode 100644 index 0000000..7240a39 --- /dev/null +++ b/internal/dockerdeploy/controlled_session_supervisor_integration_test.go @@ -0,0 +1,56 @@ +package dockerdeploy + +import ( + "context" + "os" + "os/exec" + "runtime" + "strings" + "testing" + "time" + + "github.com/omry/reploy/internal/controlledsession" +) + +func TestControlledSessionSupervisorDockerIntegration(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 supervisor Docker integration requires a supported Linux host, got %s/%s", runtime.GOOS, runtime.GOARCH) + } + ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute) + defer cancel() + image := buildControlledSessionControllerIntegrationImageV1(t, ctx) + plan := controlledSessionControllerIntegrationPlanV1(t, image, []string{"/session-channel-helper", "supervise"}) + + result, err := RunControlledSessionV1(ctx, plan, ControlledSessionRunOptionsV1{ + StartupTimeout: 30 * time.Second, TerminationGrace: 5 * time.Second, + ControllerFinalizationTimeout: 15 * time.Second, ResultAcknowledgementTimeout: 5 * time.Second, + CleanupTimeout: 15 * time.Second, + }) + if err != nil { + t.Fatal(err) + } + if result.SessionResult.Cause != controlledsession.CauseWorkloadExitV1 || + result.SessionResult.WorkloadStatus.Code == nil || *result.SessionResult.WorkloadStatus.Code != 42 || + result.SessionResult.WorkloadOutputFinalizationStatus.Kind != controlledsession.WorkloadOutputFinalizationDrainedV1 || + result.SessionResult.ControllerFinalizationStatus.Kind != controlledsession.ControllerFinalizationCompletedV1 || + result.SessionResult.CleanupStatus.Kind != controlledsession.CleanupStatusSucceededV1 || + !result.ResultDelivered || !result.ResultAcknowledged { + t.Fatalf("supervisor result = %#v", result) + } + if result.ControllerStatus.Code == nil || *result.ControllerStatus.Code != 0 || + result.DeliveryTailCleanupStatus.Kind != controlledsession.CleanupStatusSucceededV1 { + t.Fatalf("delivery-tail result = %#v", result) + } + for _, container := range []string{plan.Controller.Container, plan.Workload.Container} { + output, inspectErr := exec.CommandContext(ctx, "docker", "container", "inspect", container).CombinedOutput() + if inspectErr == nil || !strings.Contains(string(output), "No such container") { + t.Fatalf("controlled-session container %q survived cleanup: %v\n%s", container, inspectErr, output) + } + } + if _, statErr := os.Stat(plan.Channel.HostDirectory); !os.IsNotExist(statErr) { + t.Fatalf("private channel directory survived cleanup: %v", statErr) + } +} diff --git a/internal/dockerdeploy/controlled_session_supervisor_test.go b/internal/dockerdeploy/controlled_session_supervisor_test.go new file mode 100644 index 0000000..67443e4 --- /dev/null +++ b/internal/dockerdeploy/controlled_session_supervisor_test.go @@ -0,0 +1,938 @@ +package dockerdeploy + +import ( + "context" + "errors" + "io" + "sync" + "testing" + "time" + + "github.com/omry/reploy/internal/controlledsession" +) + +func TestRunControlledSessionV1OwnsNormalLifecycle(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + requests := make(chan controlledsession.RequestV1, 8) + controller := newFakeControlledSessionProcessV1() + transport := &fakeControlledSessionTransportV1{requests: requests} + transport.onEvent = func(event controlledsession.EventV1) { + switch event.Kind { + case controlledsession.EventWorkloadOutputsFinalizedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestCompleteV1} + case controlledsession.EventTerminatedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestAcknowledgeTerminatedV1} + code := 0 + controller.exit <- controlledSessionProcessResultV1{ + status: controlledsession.ProcessStatusV1{Kind: controlledsession.ProcessStatusExitedV1, Code: &code}, + } + close(requests) + } + } + workload := newFakeControlledSessionWorkloadV1([]byte("hello from workload\n"), 17) + channel := &fakeControlledSessionChannelV1{transport: transport} + + result, err := runControlledSessionV1(t.Context(), plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{ + prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { + return channel, nil + }, + prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { + return controller, nil + }, + prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { + return workload, nil + }, + now: time.Now, + }) + if err != nil { + t.Fatal(err) + } + if result.SessionResult.Cause != controlledsession.CauseWorkloadExitV1 || + result.SessionResult.WorkloadStatus.Code == nil || *result.SessionResult.WorkloadStatus.Code != 17 { + t.Fatalf("session result = %#v", result.SessionResult) + } + if !result.ResultDelivered || !result.ResultAcknowledged { + t.Fatalf("result delivery = delivered %t acknowledged %t", result.ResultDelivered, result.ResultAcknowledged) + } + if result.DeliveryTailCleanupStatus.Kind != controlledsession.CleanupStatusSucceededV1 || + result.DeliveryTailRecoveryAction != controlledsession.RecoveryNoneV1 { + t.Fatalf("delivery-tail cleanup = %#v / %q", result.DeliveryTailCleanupStatus, result.DeliveryTailRecoveryAction) + } + if !controller.cleaned || !workload.cleaned || !channel.closed { + t.Fatalf("cleanup = controller %t workload %t channel %t", controller.cleaned, workload.cleaned, channel.closed) + } + + events := transport.snapshotEvents() + if len(events) != 5 { + t.Fatalf("events = %#v", events) + } + indices := map[controlledsession.EventKindV1]int{} + for index, event := range events { + indices[event.Kind] = index + } + for _, kind := range []controlledsession.EventKindV1{ + controlledsession.EventOutputV1, + controlledsession.EventWorkloadExitV1, + controlledsession.EventTerminatingV1, + controlledsession.EventWorkloadOutputsFinalizedV1, + controlledsession.EventTerminatedV1, + } { + if _, found := indices[kind]; !found { + t.Fatalf("event %q missing from %#v", kind, events) + } + } + if indices[controlledsession.EventWorkloadExitV1] > indices[controlledsession.EventTerminatingV1] || + indices[controlledsession.EventOutputV1] > indices[controlledsession.EventWorkloadOutputsFinalizedV1] || + indices[controlledsession.EventWorkloadOutputsFinalizedV1] > indices[controlledsession.EventTerminatedV1] { + t.Fatalf("event order = %#v", events) + } + if string(events[indices[controlledsession.EventOutputV1]].Bytes) != "hello from workload\n" { + t.Fatalf("output events = %#v", events) + } +} + +func TestRunControlledSessionV1HoldsCompleteUntilOutputFinalizationPublication(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + requests := make(chan controlledsession.RequestV1, 8) + completeRead := make(chan struct{}) + requestAfterCompleteRead := make(chan struct{}) + controller := newFakeControlledSessionProcessV1() + transport := &fakeControlledSessionTransportV1{ + requests: requests, + blockEventKind: controlledsession.EventWorkloadOutputsFinalizedV1, + eventWriteBlocked: make(chan struct{}), + releaseEventWrite: make(chan struct{}), + } + var readMu sync.Mutex + readCount := 0 + transport.onRequest = func(request controlledsession.RequestV1) { + readMu.Lock() + defer readMu.Unlock() + if request.Kind == controlledsession.RequestCompleteV1 { + readCount++ + if readCount == 1 { + close(completeRead) + } + return + } + if request.Kind == controlledsession.RequestTerminateV1 && readCount == 1 { + close(requestAfterCompleteRead) + } + } + transport.onEvent = func(event controlledsession.EventV1) { + switch event.Kind { + case controlledsession.EventWorkloadOutputsFinalizedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestCompleteV1} + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestTerminateV1} + case controlledsession.EventTerminatedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestAcknowledgeTerminatedV1} + code := 0 + controller.exit <- controlledSessionProcessResultV1{ + status: controlledsession.ProcessStatusV1{Kind: controlledsession.ProcessStatusExitedV1, Code: &code}, + } + close(requests) + } + } + workload := newFakeControlledSessionWorkloadV1(nil, 0) + result := make(chan error, 1) + go func() { + _, err := runControlledSessionV1(t.Context(), plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{ + prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { + return &fakeControlledSessionChannelV1{transport: transport}, nil + }, + prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { + return controller, nil + }, + prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { + return workload, nil + }, + now: time.Now, + }) + result <- err + }() + + select { + case <-transport.eventWriteBlocked: + case <-time.After(time.Second): + t.Fatal("workload output finalization event publication did not begin") + } + select { + case <-completeRead: + case <-time.After(time.Second): + t.Fatal("controller completion request was not read during event publication") + } + select { + case <-requestAfterCompleteRead: + t.Fatal("completion was accepted before workload output finalization publication resolved") + case <-time.After(50 * time.Millisecond): + } + close(transport.releaseEventWrite) + select { + case err := <-result: + if err != nil { + t.Fatal(err) + } + case <-time.After(3 * time.Second): + t.Fatal("controlled session did not finish after output publication resolved") + } +} + +func TestRunControlledSessionV1CapsShutdownAtOutputDeadline(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + requests := make(chan controlledsession.RequestV1, 4) + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestTerminateV1} + transport := &fakeControlledSessionTransportV1{requests: requests} + transport.onEvent = func(event controlledsession.EventV1) { + switch event.Kind { + case controlledsession.EventWorkloadOutputsFinalizedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestCompleteV1} + case controlledsession.EventTerminatedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestAcknowledgeTerminatedV1} + } + } + controller := newFakeControlledSessionProcessV1() + workload := newFakeControlledSessionWorkloadV1(nil, 137) + workload.exitOnStart = false + workload.exitOnGraceful = false + options := testControlledSessionRunOptionsV1() + options.TerminationGrace = time.Hour + + started := time.Now() + _, err := runControlledSessionV1(t.Context(), plan, options, controlledSessionSupervisorBackendV1{ + prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { + return &fakeControlledSessionChannelV1{transport: transport}, nil + }, + prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { + return controller, nil + }, + prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { + return workload, nil + }, + now: func() time.Time { + return time.Now().Add(-controlledSessionOutputFinalizationTimeoutV1) + }, + }) + if err == nil { + t.Fatal("expired output deadline unexpectedly produced a successful session") + } + if elapsed := time.Since(started); elapsed > time.Second { + t.Fatalf("expired output deadline still waited %s", elapsed) + } + if !workload.forceStopped { + t.Fatal("expired output deadline did not attempt forced shutdown") + } +} + +func TestRunControlledSessionV1AppliesControllerRequests(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + requests := make(chan controlledsession.RequestV1, 8) + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestInputV1, Bytes: []byte{0, 1, 0xff}} + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestResizeV1, Columns: 132, Rows: 43} + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestTerminateV1} + transport := &fakeControlledSessionTransportV1{requests: requests} + transport.onEvent = func(event controlledsession.EventV1) { + switch event.Kind { + case controlledsession.EventWorkloadOutputsFinalizedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestCompleteV1} + case controlledsession.EventTerminatedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestAcknowledgeTerminatedV1} + } + } + controller := newFakeControlledSessionProcessV1() + workload := newFakeControlledSessionWorkloadV1(nil, 143) + workload.exitOnStart = false + + result, err := runControlledSessionV1(t.Context(), plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{ + prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { + return &fakeControlledSessionChannelV1{transport: transport}, nil + }, + prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { + return controller, nil + }, + prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { + return workload, nil + }, + now: time.Now, + }) + if err != nil { + t.Fatal(err) + } + if result.SessionResult.Cause != controlledsession.CauseControllerTerminateV1 { + t.Fatalf("cause = %q", result.SessionResult.Cause) + } + if !workload.gracefulStopped { + t.Fatal("workload did not receive graceful stop") + } + if got := workload.snapshotInput(); string(got) != string([]byte{0, 1, 0xff}) { + t.Fatalf("input = %v", got) + } + if workload.columns != 132 || workload.rows != 43 { + t.Fatalf("dimensions = %dx%d", workload.columns, workload.rows) + } +} + +func TestRunControlledSessionV1CleansPreparedResourcesAfterStartupFailure(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + controller := newFakeControlledSessionProcessV1() + workload := newFakeControlledSessionWorkloadV1(nil, 0) + startupErr := errors.New("workload start failed") + workload.startErr = startupErr + requests := make(chan controlledsession.RequestV1, 1) + transport := &fakeControlledSessionTransportV1{requests: requests} + transport.onEvent = func(event controlledsession.EventV1) { + if event.Kind == controlledsession.EventTerminatedV1 { + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestAcknowledgeTerminatedV1} + } + } + channel := &fakeControlledSessionChannelV1{transport: transport} + + result, err := runControlledSessionV1(t.Context(), plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{ + prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { return channel, nil }, + prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { + return controller, nil + }, + prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { + return workload, nil + }, + now: time.Now, + }) + if !errors.Is(err, startupErr) { + t.Fatalf("error = %v", err) + } + if result.SessionResult.Cause != controlledsession.CauseStartupFailureV1 { + t.Fatalf("result = %#v", result.SessionResult) + } + if !result.ResultDelivered || !result.ResultAcknowledged { + t.Fatalf("startup result delivery = delivered %t acknowledged %t", result.ResultDelivered, result.ResultAcknowledged) + } + if !controller.cleaned || !workload.cleaned || !channel.closed { + t.Fatalf("cleanup = controller %t workload %t channel %t", controller.cleaned, workload.cleaned, channel.closed) + } +} + +func TestRunControlledSessionV1RecordsHostCancellationDuringStartup(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + channel := &fakeControlledSessionChannelV1{} + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + result, err := runControlledSessionV1(ctx, plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{ + prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { + return channel, nil + }, + prepareController: func(ctx context.Context, _ ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { + return nil, ctx.Err() + }, + prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { + t.Fatal("workload preparation continued after host cancellation") + return nil, nil + }, + now: time.Now, + }) + if !errors.Is(err, context.Canceled) { + t.Fatalf("error = %v", err) + } + if result.SessionResult.Cause != controlledsession.CauseHostCancelV1 || + result.SessionResult.ControllerFinalizationStatus.Kind != controlledsession.ControllerFinalizationStartupFailedV1 || + result.ResultDelivered || !channel.closed { + t.Fatalf("startup cancellation result = %#v, channel closed = %t", result, channel.closed) + } +} + +func TestRunControlledSessionV1DrainsPartiallyStartedWorkloadAfterStartupFailure(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + controller := newFakeControlledSessionProcessV1() + workload := newFakeControlledSessionWorkloadV1([]byte("output before initial resize failed\n"), 143) + startupErr := errors.New("initial terminal resize failed") + workload.startPartially = true + workload.startErr = startupErr + requests := make(chan controlledsession.RequestV1, 1) + transport := &fakeControlledSessionTransportV1{requests: requests} + transport.onEvent = func(event controlledsession.EventV1) { + if event.Kind == controlledsession.EventTerminatedV1 { + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestAcknowledgeTerminatedV1} + } + } + + result, err := runControlledSessionV1(t.Context(), plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{ + prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { + return &fakeControlledSessionChannelV1{transport: transport}, nil + }, + prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { + return controller, nil + }, + prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { + return workload, nil + }, + now: time.Now, + }) + if !errors.Is(err, startupErr) { + t.Fatalf("error = %v", err) + } + if result.SessionResult.Cause != controlledsession.CauseStartupFailureV1 || + result.SessionResult.WorkloadStatus.Code == nil || *result.SessionResult.WorkloadStatus.Code != 143 || + result.SessionResult.WorkloadOutputFinalizationStatus.Kind != controlledsession.WorkloadOutputFinalizationDrainedV1 { + t.Fatalf("session result = %#v", result.SessionResult) + } + if !workload.gracefulStopped || !workload.cleaned { + t.Fatalf("workload teardown = graceful %t cleaned %t", workload.gracefulStopped, workload.cleaned) + } + events := transport.snapshotEvents() + indices := map[controlledsession.EventKindV1]int{} + for index, event := range events { + indices[event.Kind] = index + } + for _, kind := range []controlledsession.EventKindV1{ + controlledsession.EventOutputV1, + controlledsession.EventWorkloadExitV1, + controlledsession.EventTerminatingV1, + controlledsession.EventWorkloadOutputsFinalizedV1, + controlledsession.EventTerminatedV1, + } { + if _, found := indices[kind]; !found { + t.Fatalf("event %q missing from %#v", kind, events) + } + } + if indices[controlledsession.EventOutputV1] > indices[controlledsession.EventWorkloadOutputsFinalizedV1] || + indices[controlledsession.EventWorkloadOutputsFinalizedV1] > indices[controlledsession.EventTerminatedV1] { + t.Fatalf("event order = %#v", events) + } +} + +func TestRunControlledSessionV1FailsClosedAfterRuntimeObservationLoss(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + requests := make(chan controlledsession.RequestV1, 4) + transport := &fakeControlledSessionTransportV1{requests: requests} + transport.onEvent = func(event controlledsession.EventV1) { + switch event.Kind { + case controlledsession.EventWorkloadOutputsFinalizedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestCompleteV1} + case controlledsession.EventTerminatedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestAcknowledgeTerminatedV1} + } + } + controller := newFakeControlledSessionProcessV1() + workload := newFakeControlledSessionWorkloadV1(nil, 0) + observationErr := errors.New("runtime observation unavailable") + workload.waitErr = observationErr + workload.exitOnGraceful = false + + result, err := runControlledSessionV1(t.Context(), plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{ + prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { + return &fakeControlledSessionChannelV1{transport: transport}, nil + }, + prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { + return controller, nil + }, + prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { + return workload, nil + }, + now: time.Now, + }) + if !errors.Is(err, observationErr) { + t.Fatalf("error = %v", err) + } + if result.SessionResult.Cause != controlledsession.CauseRuntimeObservationLostV1 || + result.SessionResult.RuntimeObservationStatus.Kind != controlledsession.RuntimeObservationLostV1 || + result.SessionResult.WorkloadStatus.Kind != controlledsession.ProcessStatusUnavailableV1 || + result.SessionResult.WorkloadOutputFinalizationStatus.Kind != controlledsession.WorkloadOutputFinalizationFailedV1 { + t.Fatalf("session result = %#v", result.SessionResult) + } + if !workload.gracefulStopped || !workload.forceStopped || !workload.cleaned { + t.Fatalf("workload teardown = graceful %t force %t cleaned %t", workload.gracefulStopped, workload.forceStopped, workload.cleaned) + } +} + +func TestRunControlledSessionV1StopsWorkloadAfterPTYReadFailure(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + requests := make(chan controlledsession.RequestV1, 4) + transport := &fakeControlledSessionTransportV1{requests: requests} + transport.onEvent = func(event controlledsession.EventV1) { + switch event.Kind { + case controlledsession.EventWorkloadOutputsFinalizedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestCompleteV1} + case controlledsession.EventTerminatedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestAcknowledgeTerminatedV1} + } + } + controller := newFakeControlledSessionProcessV1() + workload := newFakeControlledSessionWorkloadV1(nil, 143) + workload.exitOnStart = false + readErr := errors.New("PTY attachment failed") + if err := workload.writer.CloseWithError(readErr); err != nil { + t.Fatal(err) + } + + result, err := runControlledSessionV1(t.Context(), plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{ + prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { + return &fakeControlledSessionChannelV1{transport: transport}, nil + }, + prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { + return controller, nil + }, + prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { + return workload, nil + }, + now: time.Now, + }) + if !errors.Is(err, readErr) { + t.Fatalf("error = %v", err) + } + if result.SessionResult.Cause != controlledsession.CauseRuntimeObservationLostV1 || + result.SessionResult.WorkloadOutputFinalizationStatus.Kind != controlledsession.WorkloadOutputFinalizationFailedV1 || + !result.ResultDelivered || !workload.gracefulStopped || !workload.cleaned { + t.Fatalf("output-loss result = %#v, workload = %#v", result, workload) + } +} + +func TestRunControlledSessionV1StopsWorkloadAfterOutputDeliveryFailure(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + deliveryErr := errors.New("controller event transport failed") + transport := &fakeControlledSessionTransportV1{ + requests: make(chan controlledsession.RequestV1), + writeErr: deliveryErr, + } + controller := newFakeControlledSessionProcessV1() + workload := newFakeControlledSessionWorkloadV1(nil, 143) + workload.exitOnStart = false + writeDone := make(chan error, 1) + go func() { + _, err := workload.writer.Write([]byte("undeliverable output")) + writeDone <- err + }() + + result, err := runControlledSessionV1(t.Context(), plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{ + prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { + return &fakeControlledSessionChannelV1{transport: transport}, nil + }, + prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { + return controller, nil + }, + prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { + return workload, nil + }, + now: time.Now, + }) + if !errors.Is(err, deliveryErr) { + t.Fatalf("error = %v", err) + } + if writeErr := <-writeDone; writeErr != nil { + t.Fatalf("write workload output: %v", writeErr) + } + if result.SessionResult.Cause != controlledsession.CauseControllerLostV1 || + result.SessionResult.WorkloadOutputFinalizationStatus.Kind != controlledsession.WorkloadOutputFinalizationFailedV1 || + result.ResultDelivered || !workload.gracefulStopped || !workload.cleaned { + t.Fatalf("delivery-loss result = %#v, workload = %#v", result, workload) + } +} + +func TestRunControlledSessionV1StopsWorkloadAfterHostCancellation(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + requests := make(chan controlledsession.RequestV1, 4) + transport := &fakeControlledSessionTransportV1{requests: requests} + transport.onEvent = func(event controlledsession.EventV1) { + switch event.Kind { + case controlledsession.EventWorkloadOutputsFinalizedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestCompleteV1} + case controlledsession.EventTerminatedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestAcknowledgeTerminatedV1} + } + } + controller := newFakeControlledSessionProcessV1() + workload := newFakeControlledSessionWorkloadV1(nil, 143) + workload.exitOnStart = false + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + result, err := runControlledSessionV1(ctx, plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{ + prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { + return &fakeControlledSessionChannelV1{transport: transport}, nil + }, + prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { + return controller, nil + }, + prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { + return workload, nil + }, + now: time.Now, + }) + if err != nil { + t.Fatal(err) + } + if result.SessionResult.Cause != controlledsession.CauseHostCancelV1 || !workload.gracefulStopped || !workload.cleaned { + t.Fatalf("host cancellation result = %#v, workload = %#v", result, workload) + } +} + +func TestRunControlledSessionV1RecordsLatePTYFailureWithoutReplacingHostCancellation(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + requests := make(chan controlledsession.RequestV1, 4) + transport := &fakeControlledSessionTransportV1{requests: requests} + transport.onEvent = func(event controlledsession.EventV1) { + switch event.Kind { + case controlledsession.EventWorkloadOutputsFinalizedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestCompleteV1} + case controlledsession.EventTerminatedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestAcknowledgeTerminatedV1} + } + } + controller := newFakeControlledSessionProcessV1() + workload := newFakeControlledSessionWorkloadV1(nil, 143) + workload.exitOnStart = false + lateReadErr := errors.New("PTY failed during teardown") + workload.gracefulOutputErr = lateReadErr + ctx, cancel := context.WithCancel(context.Background()) + cancel() + + result, err := runControlledSessionV1(ctx, plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{ + prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { + return &fakeControlledSessionChannelV1{transport: transport}, nil + }, + prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { + return controller, nil + }, + prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { + return workload, nil + }, + now: time.Now, + }) + if !errors.Is(err, lateReadErr) { + t.Fatalf("error = %v", err) + } + if result.SessionResult.Cause != controlledsession.CauseHostCancelV1 || + result.SessionResult.RuntimeObservationStatus.Kind != controlledsession.RuntimeObservationLostV1 || + result.SessionResult.WorkloadOutputFinalizationStatus.Kind != controlledsession.WorkloadOutputFinalizationFailedV1 || + !result.ResultDelivered || !result.ResultAcknowledged { + t.Fatalf("late output failure result = %#v", result) + } +} + +func TestRunControlledSessionV1CancelsBlockedInputBeforeStoppingWorkload(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + requests := make(chan controlledsession.RequestV1, 4) + transport := &fakeControlledSessionTransportV1{requests: requests} + transport.onEvent = func(event controlledsession.EventV1) { + switch event.Kind { + case controlledsession.EventWorkloadOutputsFinalizedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestCompleteV1} + case controlledsession.EventTerminatedV1: + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestAcknowledgeTerminatedV1} + } + } + controller := newFakeControlledSessionProcessV1() + workload := newFakeControlledSessionWorkloadV1(nil, 143) + workload.exitOnStart = false + workload.blockInput = true + ctx, cancel := context.WithCancel(context.Background()) + resultDone := make(chan struct { + result ControlledSessionRunResultV1 + err error + }, 1) + go func() { + result, err := runControlledSessionV1(ctx, plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{ + prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { + return &fakeControlledSessionChannelV1{transport: transport}, nil + }, + prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { + return controller, nil + }, + prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { + return workload, nil + }, + now: time.Now, + }) + resultDone <- struct { + result ControlledSessionRunResultV1 + err error + }{result: result, err: err} + }() + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestInputV1, Bytes: []byte("blocked")} + select { + case <-workload.inputStarted: + case <-time.After(2 * time.Second): + t.Fatal("input request did not reach workload") + } + cancel() + select { + case invocation := <-resultDone: + if invocation.err != nil { + t.Fatal(invocation.err) + } + if invocation.result.SessionResult.Cause != controlledsession.CauseHostCancelV1 || + !invocation.result.ResultDelivered || !invocation.result.ResultAcknowledged || + !workload.gracefulStopped || !workload.cleaned { + t.Fatalf("host cancellation result = %#v, workload = %#v", invocation.result, workload) + } + case <-time.After(2 * time.Second): + t.Fatal("controlled session did not cancel blocked input before workload teardown") + } +} + +func TestRunControlledSessionV1FailsClosedAfterControllerExit(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + controller := newFakeControlledSessionProcessV1() + controllerCode := 9 + controller.exit <- controlledSessionProcessResultV1{ + status: controlledsession.ProcessStatusV1{Kind: controlledsession.ProcessStatusExitedV1, Code: &controllerCode}, + } + workload := newFakeControlledSessionWorkloadV1(nil, 143) + workload.exitOnStart = false + transport := &fakeControlledSessionTransportV1{requests: make(chan controlledsession.RequestV1)} + + result, err := runControlledSessionV1(t.Context(), plan, testControlledSessionRunOptionsV1(), controlledSessionSupervisorBackendV1{ + prepareChannel: func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { + return &fakeControlledSessionChannelV1{transport: transport}, nil + }, + prepareController: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { + return controller, nil + }, + prepareWorkload: func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { + return workload, nil + }, + now: time.Now, + }) + if err != nil { + t.Fatal(err) + } + if result.SessionResult.Cause != controlledsession.CauseControllerLostV1 || + result.SessionResult.ControllerFinalizationStatus.Kind != controlledsession.ControllerFinalizationLostV1 || + result.ResultDelivered || !workload.gracefulStopped || !workload.cleaned { + t.Fatalf("controller-loss result = %#v, workload = %#v", result, workload) + } +} + +func testControlledSessionRunOptionsV1() ControlledSessionRunOptionsV1 { + return ControlledSessionRunOptionsV1{ + StartupTimeout: 2 * time.Second, TerminationGrace: 100 * time.Millisecond, + ControllerFinalizationTimeout: 2 * time.Second, ResultAcknowledgementTimeout: 2 * time.Second, + CleanupTimeout: 2 * time.Second, + } +} + +type fakeControlledSessionTransportV1 struct { + requests chan controlledsession.RequestV1 + mu sync.Mutex + events []controlledsession.EventV1 + onRequest func(controlledsession.RequestV1) + onEvent func(controlledsession.EventV1) + writeErr error + blockEventKind controlledsession.EventKindV1 + eventWriteBlocked chan struct{} + releaseEventWrite chan struct{} +} + +func (transport *fakeControlledSessionTransportV1) ReadRequest(ctx context.Context) (controlledsession.RequestV1, error) { + select { + case request, ok := <-transport.requests: + if !ok { + return controlledsession.RequestV1{}, io.EOF + } + if transport.onRequest != nil { + transport.onRequest(request) + } + return request, nil + case <-ctx.Done(): + return controlledsession.RequestV1{}, ctx.Err() + } +} + +func (transport *fakeControlledSessionTransportV1) WriteEvent(ctx context.Context, event controlledsession.EventV1) error { + if transport.writeErr != nil { + return transport.writeErr + } + event.Bytes = append([]byte(nil), event.Bytes...) + transport.mu.Lock() + transport.events = append(transport.events, event) + transport.mu.Unlock() + if transport.onEvent != nil { + transport.onEvent(event) + } + if event.Kind == transport.blockEventKind { + close(transport.eventWriteBlocked) + select { + case <-transport.releaseEventWrite: + case <-ctx.Done(): + return ctx.Err() + } + } + return nil +} + +func (transport *fakeControlledSessionTransportV1) snapshotEvents() []controlledsession.EventV1 { + transport.mu.Lock() + defer transport.mu.Unlock() + return append([]controlledsession.EventV1(nil), transport.events...) +} + +type fakeControlledSessionChannelV1 struct { + transport controlledsession.ControllerTransportV1 + closed bool +} + +func (channel *fakeControlledSessionChannelV1) Claim(context.Context) (controlledsession.ControllerTransportV1, error) { + return channel.transport, nil +} + +func (channel *fakeControlledSessionChannelV1) Close() error { + channel.closed = true + return nil +} + +type fakeControlledSessionProcessV1 struct { + exit chan controlledSessionProcessResultV1 + started bool + gracefulStopped bool + forceStopped bool + cleaned bool +} + +func newFakeControlledSessionProcessV1() *fakeControlledSessionProcessV1 { + return &fakeControlledSessionProcessV1{exit: make(chan controlledSessionProcessResultV1, 1)} +} + +func (process *fakeControlledSessionProcessV1) Start(context.Context) error { + process.started = true + return nil +} + +func (process *fakeControlledSessionProcessV1) Wait(ctx context.Context) (controlledsession.ProcessStatusV1, error) { + select { + case result := <-process.exit: + return result.status, result.err + case <-ctx.Done(): + return controlledsession.ProcessStatusV1{}, ctx.Err() + } +} + +func (process *fakeControlledSessionProcessV1) RequestGracefulStop(context.Context) error { + process.gracefulStopped = true + code := 0 + process.exit <- controlledSessionProcessResultV1{status: controlledsession.ProcessStatusV1{Kind: controlledsession.ProcessStatusExitedV1, Code: &code}} + return nil +} + +func (process *fakeControlledSessionProcessV1) ForceStop(context.Context) error { + process.forceStopped = true + return nil +} + +func (process *fakeControlledSessionProcessV1) Cleanup(context.Context) error { + process.cleaned = true + return nil +} + +type fakeControlledSessionWorkloadV1 struct { + *fakeControlledSessionProcessV1 + reader *io.PipeReader + writer *io.PipeWriter + output []byte + exitCode int + exitOnStart bool + exitOnGraceful bool + gracefulOutputErr error + startPartially bool + startErr error + waitErr error + closed bool + inputMu sync.Mutex + input []byte + blockInput bool + inputStarted chan struct{} + inputStartOnce sync.Once + columns uint32 + rows uint32 +} + +func newFakeControlledSessionWorkloadV1(output []byte, exitCode int) *fakeControlledSessionWorkloadV1 { + reader, writer := io.Pipe() + return &fakeControlledSessionWorkloadV1{ + fakeControlledSessionProcessV1: newFakeControlledSessionProcessV1(), + reader: reader, writer: writer, output: append([]byte(nil), output...), exitCode: exitCode, exitOnStart: true, exitOnGraceful: true, + inputStarted: make(chan struct{}), + } +} + +func (workload *fakeControlledSessionWorkloadV1) Output() (io.ReadCloser, error) { + return workload.reader, nil +} + +func (workload *fakeControlledSessionWorkloadV1) Start(context.Context) error { + if workload.startErr != nil { + workload.started = workload.startPartially + if workload.startPartially && len(workload.output) > 0 { + _, _ = workload.writer.Write(workload.output) + } + return workload.startErr + } + workload.started = true + if workload.exitOnStart { + go func() { + _, _ = workload.writer.Write(workload.output) + _ = workload.writer.Close() + if workload.waitErr != nil { + workload.exit <- controlledSessionProcessResultV1{ + status: controlledsession.ProcessStatusV1{Kind: controlledsession.ProcessStatusUnavailableV1, Reason: "workload runtime observation was lost"}, + err: workload.waitErr, + } + return + } + code := workload.exitCode + workload.exit <- controlledSessionProcessResultV1{status: controlledsession.ProcessStatusV1{Kind: controlledsession.ProcessStatusExitedV1, Code: &code}} + }() + } + return nil +} + +func (workload *fakeControlledSessionWorkloadV1) Started() bool { + return workload.started +} + +func (workload *fakeControlledSessionWorkloadV1) RequestGracefulStop(context.Context) error { + workload.gracefulStopped = true + if workload.gracefulOutputErr != nil { + _ = workload.writer.CloseWithError(workload.gracefulOutputErr) + } else { + _ = workload.writer.Close() + } + if !workload.exitOnGraceful { + return nil + } + code := workload.exitCode + workload.exit <- controlledSessionProcessResultV1{status: controlledsession.ProcessStatusV1{Kind: controlledsession.ProcessStatusExitedV1, Code: &code}} + return nil +} + +func (workload *fakeControlledSessionWorkloadV1) ForceStop(context.Context) error { + workload.forceStopped = true + code := workload.exitCode + workload.exit <- controlledSessionProcessResultV1{status: controlledsession.ProcessStatusV1{Kind: controlledsession.ProcessStatusExitedV1, Code: &code}} + return nil +} + +func (workload *fakeControlledSessionWorkloadV1) WriteInput(ctx context.Context, input []byte) error { + workload.inputStartOnce.Do(func() { close(workload.inputStarted) }) + if workload.blockInput { + <-ctx.Done() + return ctx.Err() + } + workload.inputMu.Lock() + defer workload.inputMu.Unlock() + workload.input = append(workload.input, input...) + return nil +} + +func (workload *fakeControlledSessionWorkloadV1) Resize(_ context.Context, columns uint32, rows uint32) error { + workload.columns, workload.rows = columns, rows + return nil +} + +func (workload *fakeControlledSessionWorkloadV1) Close() error { + workload.closed = true + return workload.reader.Close() +} + +func (workload *fakeControlledSessionWorkloadV1) snapshotInput() []byte { + workload.inputMu.Lock() + defer workload.inputMu.Unlock() + return append([]byte(nil), workload.input...) +} diff --git a/internal/dockerdeploy/controlled_session_workload_pty.go b/internal/dockerdeploy/controlled_session_workload_pty.go index 266e884..f0fea71 100644 --- a/internal/dockerdeploy/controlled_session_workload_pty.go +++ b/internal/dockerdeploy/controlled_session_workload_pty.go @@ -37,14 +37,16 @@ type dockerWorkloadPTYWaitResultV1 struct { // controlled-session host operation. Closing it closes only the PTY // attachment; container cleanup remains the lifecycle supervisor's job. type DockerWorkloadPTYV1 struct { - plan ControlledSessionContainerPlanV1 - backend dockerWorkloadPTYBackendV1 - attachment dockerPTYAttachmentV1 + plan ControlledSessionContainerPlanV1 + containerID string + backend dockerWorkloadPTYBackendV1 + attachment dockerPTYAttachmentV1 operationMu sync.Mutex stateMu sync.Mutex started bool outputTaken bool + cleaned bool waitDone chan struct{} waitResult dockerWorkloadPTYWaitResultV1 @@ -85,19 +87,28 @@ func prepareDockerWorkloadPTYV1( ctx = context.Background() } create := controlledSessionCommandSpecV1(plan.Create) - if err := backend.run(create, RunOptions{Context: ctx}); err != nil { + var createOutput bytes.Buffer + var createErrorOutput bytes.Buffer + if err := backend.run(create, RunOptions{Context: ctx, Stdout: &createOutput, Stderr: &createErrorOutput}); err != nil { + if output := trimmedCommandOutput(createErrorOutput.String()); output != "" { + err = fmt.Errorf("%w\ncommand output:\n%s", err, output) + } return nil, fmt.Errorf("create controlled-session workload container %q: %w", plan.Container, err) } - attachment, err := backend.attach(ctx, create, plan.Container, defaultDockerPreflightTimeout) + containerID, err := parseDockerContainerIDV1(createOutput.String()) + if err != nil { + return nil, fmt.Errorf("create controlled-session workload container %q: %w; refusing name-based cleanup because the created container identity is unknown", plan.Container, err) + } + attachment, err := backend.attach(ctx, create, containerID, defaultDockerPreflightTimeout) if err != nil { attachErr := fmt.Errorf("attach controlled-session workload PTY for container %q before start: %w", plan.Container, err) - if cleanupErr := rollbackControlledSessionWorkloadContainerV1(backend, plan); cleanupErr != nil { + if cleanupErr := rollbackControlledSessionWorkloadContainerV1(backend, plan, containerID); cleanupErr != nil { return nil, errors.Join(attachErr, fmt.Errorf("remove inert controlled-session workload container %q after attach failure: %w", plan.Container, cleanupErr)) } return nil, attachErr } return &DockerWorkloadPTYV1{ - plan: plan, backend: backend, attachment: attachment, waitDone: make(chan struct{}), + plan: plan, containerID: containerID, backend: backend, attachment: attachment, waitDone: make(chan struct{}), }, nil } @@ -113,6 +124,15 @@ func (workload *DockerWorkloadPTYV1) Output() (io.ReadCloser, error) { return workload.attachment, nil } +// Started reports whether Docker accepted the workload start. Start can fail +// after this becomes true—for example, while applying the initial PTY size—so +// the supervisor uses it to select the full teardown and output barrier. +func (workload *DockerWorkloadPTYV1) Started() bool { + workload.stateMu.Lock() + defer workload.stateMu.Unlock() + return workload.started +} + // Start starts the already-attached inert container, begins independent exit // observation, and applies the immutable initial terminal dimensions before // returning success. Docker does not permit resizing a created container, so @@ -136,10 +156,16 @@ func (workload *DockerWorkloadPTYV1) Start(ctx context.Context) error { if ctx == nil { ctx = context.Background() } - if err := workload.backend.run(controlledSessionCommandSpecV1(workload.plan.Start), RunOptions{Context: ctx}); err != nil { + start := CommandSpec{Name: workload.plan.Start.Name, Args: []string{"start", workload.containerID}} + if err := workload.backend.run(start, RunOptions{Context: ctx}); err != nil { startErr := fmt.Errorf("start attached controlled-session workload container %q: %w", workload.plan.Container, err) closeErr := workload.Close() - cleanupErr := rollbackControlledSessionWorkloadContainerV1(workload.backend, workload.plan) + cleanupErr := rollbackControlledSessionWorkloadContainerV1(workload.backend, workload.plan, workload.containerID) + if cleanupErr == nil { + workload.stateMu.Lock() + workload.cleaned = true + workload.stateMu.Unlock() + } if closeErr != nil { closeErr = fmt.Errorf("close controlled-session workload PTY after start failure: %w", closeErr) } @@ -198,7 +224,7 @@ func (workload *DockerWorkloadPTYV1) resizeLocked(ctx context.Context, columns u if err := workload.backend.resize( ctx, controlledSessionCommandSpecV1(workload.plan.Start), - workload.plan.Container, + workload.containerID, columns, rows, defaultDockerPreflightTimeout, @@ -230,7 +256,7 @@ func (workload *DockerWorkloadPTYV1) signal(ctx context.Context, signal string, ctx = context.Background() } command := CommandSpec{Name: workload.plan.Start.Name, Args: []string{ - "kill", "--signal", signal, workload.plan.Container, + "kill", "--signal", signal, workload.containerID, }} if err := workload.backend.run(command, RunOptions{Context: ctx}); err != nil { return fmt.Errorf("%s controlled-session workload container %q: %w", action, workload.plan.Container, err) @@ -258,8 +284,34 @@ func (workload *DockerWorkloadPTYV1) Wait(ctx context.Context) (controlledsessio } } +// Cleanup force-removes the exact planned workload container. Successful +// cleanup is idempotent within this adapter; a failed attempt may be retried. +// The lifecycle supervisor must finalize or close PTY output before calling it. +func (workload *DockerWorkloadPTYV1) Cleanup(ctx context.Context) error { + workload.operationMu.Lock() + defer workload.operationMu.Unlock() + + workload.stateMu.Lock() + if workload.cleaned { + workload.stateMu.Unlock() + return nil + } + workload.stateMu.Unlock() + if ctx == nil { + ctx = context.Background() + } + cleanup := CommandSpec{Name: workload.plan.Cleanup.Name, Args: []string{"container", "rm", "--force", workload.containerID}} + if err := workload.backend.run(cleanup, RunOptions{Context: ctx}); err != nil { + return fmt.Errorf("remove controlled-session workload container %q: %w", workload.plan.Container, err) + } + workload.stateMu.Lock() + workload.cleaned = true + workload.stateMu.Unlock() + return nil +} + func (workload *DockerWorkloadPTYV1) observeExit() { - code, err := workload.backend.observe(context.Background(), controlledSessionCommandSpecV1(workload.plan.Start), workload.plan.Container) + code, err := workload.backend.observe(context.Background(), controlledSessionCommandSpecV1(workload.plan.Start), workload.containerID) if err != nil { workload.waitResult = dockerWorkloadPTYWaitResultV1{ status: controlledsession.ProcessStatusV1{ @@ -282,6 +334,9 @@ func (workload *DockerWorkloadPTYV1) requireStarted() error { if !workload.started { return fmt.Errorf("controlled-session workload container %q is not started", workload.plan.Container) } + if workload.cleaned { + return fmt.Errorf("controlled-session workload container %q is already cleaned", workload.plan.Container) + } return nil } @@ -300,10 +355,12 @@ func controlledSessionCommandSpecV1(command ControlledSessionDockerCommandV1) Co func rollbackControlledSessionWorkloadContainerV1( backend dockerWorkloadPTYBackendV1, plan ControlledSessionContainerPlanV1, + containerID string, ) error { cleanupCtx, cancel := context.WithTimeout(context.Background(), defaultDockerPreflightTimeout) defer cancel() - return backend.run(controlledSessionCommandSpecV1(plan.Cleanup), RunOptions{Context: cleanupCtx}) + cleanup := CommandSpec{Name: plan.Cleanup.Name, Args: []string{"container", "rm", "--force", containerID}} + return backend.run(cleanup, RunOptions{Context: cleanupCtx}) } func observeDockerContainerExitV1(ctx context.Context, docker CommandSpec, container string) (int, error) { diff --git a/internal/dockerdeploy/controlled_session_workload_pty_test.go b/internal/dockerdeploy/controlled_session_workload_pty_test.go index 635237d..f949801 100644 --- a/internal/dockerdeploy/controlled_session_workload_pty_test.go +++ b/internal/dockerdeploy/controlled_session_workload_pty_test.go @@ -16,6 +16,8 @@ import ( "github.com/omry/reploy/internal/controlledsession" ) +const dockerWorkloadTestContainerIDV1 = "bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + type fakeDockerPTYAttachmentV1 struct { mu sync.Mutex input []byte @@ -63,8 +65,9 @@ func TestDockerWorkloadPTYV1OrdersAttachStartResizeAndExactOperations(t *testing } exit := make(chan int, 1) backend := dockerWorkloadPTYBackendV1{ - run: func(spec CommandSpec, _ RunOptions) error { + run: func(spec CommandSpec, options RunOptions) error { record(strings.Join(spec.Args, " ")) + writeDockerWorkloadCreateIDV1(spec, options, plan) return nil }, attach: func(_ context.Context, _ CommandSpec, container string, _ time.Duration) (dockerPTYAttachmentV1, error) { @@ -99,6 +102,9 @@ func TestDockerWorkloadPTYV1OrdersAttachStartResizeAndExactOperations(t *testing if err := workload.Start(t.Context()); err != nil { t.Fatal(err) } + if !workload.Started() { + t.Fatal("successful workload start was not recorded") + } if err := workload.WriteInput(t.Context(), []byte{0, 1, 2, 0xff}); err != nil { t.Fatal(err) } @@ -119,6 +125,15 @@ func TestDockerWorkloadPTYV1OrdersAttachStartResizeAndExactOperations(t *testing if status.Kind != controlledsession.ProcessStatusExitedV1 || status.Code == nil || *status.Code != 42 { t.Fatalf("wait status = %#v", status) } + if err := workload.Close(); err != nil { + t.Fatal(err) + } + if err := workload.Cleanup(t.Context()); err != nil { + t.Fatal(err) + } + if err := workload.Cleanup(t.Context()); err != nil { + t.Fatalf("idempotent cleanup = %v", err) + } if !reflect.DeepEqual(attachment.input, []byte{0, 1, 2, 0xff}) { t.Fatalf("PTY input = %v", attachment.input) } @@ -128,19 +143,20 @@ func TestDockerWorkloadPTYV1OrdersAttachStartResizeAndExactOperations(t *testing actionsMu.Lock() got := append([]string(nil), actions...) actionsMu.Unlock() - if len(got) >= 7 { + if len(got) >= 8 { wantPrefix := []string{ strings.Join(plan.Create.Args, " "), - "attach " + plan.Container, - strings.Join(plan.Start.Args, " "), + "attach " + dockerWorkloadTestContainerIDV1, + "start " + dockerWorkloadTestContainerIDV1, } if !reflect.DeepEqual(got[:3], wantPrefix) { t.Fatalf("initial actions = %#v, want prefix %#v", got, wantPrefix) } - if !slicesContainStringV1(got, "resize 80x24 "+plan.Container) || - !slicesContainStringV1(got, "resize 132x43 "+plan.Container) || - !slicesContainStringV1(got, "kill --signal TERM "+plan.Container) || - !slicesContainStringV1(got, "kill --signal KILL "+plan.Container) { + if !slicesContainStringV1(got, "resize 80x24 "+dockerWorkloadTestContainerIDV1) || + !slicesContainStringV1(got, "resize 132x43 "+dockerWorkloadTestContainerIDV1) || + !slicesContainStringV1(got, "kill --signal TERM "+dockerWorkloadTestContainerIDV1) || + !slicesContainStringV1(got, "kill --signal KILL "+dockerWorkloadTestContainerIDV1) || + !slicesContainStringV1(got, "container rm --force "+dockerWorkloadTestContainerIDV1) { t.Fatalf("lifecycle actions = %#v", got) } break @@ -156,8 +172,9 @@ func TestPrepareDockerWorkloadPTYV1RollsBackInertContainerAfterAttachFailure(t * plan := controlledSessionWorkloadPlanFixtureV1(t) runs := []CommandSpec{} backend := dockerWorkloadPTYBackendV1{ - run: func(spec CommandSpec, _ RunOptions) error { + run: func(spec CommandSpec, options RunOptions) error { runs = append(runs, spec) + writeDockerWorkloadCreateIDV1(spec, options, plan) return nil }, attach: func(context.Context, CommandSpec, string, time.Duration) (dockerPTYAttachmentV1, error) { @@ -170,7 +187,8 @@ func TestPrepareDockerWorkloadPTYV1RollsBackInertContainerAfterAttachFailure(t * if err == nil || !strings.Contains(err.Error(), "before start") || !strings.Contains(err.Error(), "attach refused") { t.Fatalf("attach error = %v", err) } - if len(runs) != 2 || !reflect.DeepEqual(runs[0].Args, plan.Create.Args) || !reflect.DeepEqual(runs[1].Args, plan.Cleanup.Args) { + if len(runs) != 2 || !reflect.DeepEqual(runs[0].Args, plan.Create.Args) || + !reflect.DeepEqual(runs[1].Args, []string{"container", "rm", "--force", dockerWorkloadTestContainerIDV1}) { t.Fatalf("rollback commands = %#v", runs) } } @@ -181,9 +199,10 @@ func TestDockerWorkloadPTYV1RollsBackAmbiguousStartFailure(t *testing.T) { runs := []CommandSpec{} observed := false backend := dockerWorkloadPTYBackendV1{ - run: func(spec CommandSpec, _ RunOptions) error { + run: func(spec CommandSpec, options RunOptions) error { runs = append(runs, spec) - if reflect.DeepEqual(spec.Args, plan.Start.Args) { + writeDockerWorkloadCreateIDV1(spec, options, plan) + if reflect.DeepEqual(spec.Args, []string{"start", dockerWorkloadTestContainerIDV1}) { return errors.New("start response was lost") } return nil @@ -209,7 +228,8 @@ func TestDockerWorkloadPTYV1RollsBackAmbiguousStartFailure(t *testing.T) { t.Fatalf("ambiguous start error = %v", err) } if len(runs) != 3 || !reflect.DeepEqual(runs[0].Args, plan.Create.Args) || - !reflect.DeepEqual(runs[1].Args, plan.Start.Args) || !reflect.DeepEqual(runs[2].Args, plan.Cleanup.Args) { + !reflect.DeepEqual(runs[1].Args, []string{"start", dockerWorkloadTestContainerIDV1}) || + !reflect.DeepEqual(runs[2].Args, []string{"container", "rm", "--force", dockerWorkloadTestContainerIDV1}) { t.Fatalf("start rollback commands = %#v", runs) } attachment.mu.Lock() @@ -218,13 +238,54 @@ func TestDockerWorkloadPTYV1RollsBackAmbiguousStartFailure(t *testing.T) { if !closed || observed { t.Fatalf("start rollback closed=%t observed=%t", closed, observed) } + if workload.Started() { + t.Fatal("failed Docker start was recorded as started") + } +} + +func TestDockerWorkloadPTYV1ReportsPartialStartAfterInitialResizeFailure(t *testing.T) { + plan := controlledSessionWorkloadPlanFixtureV1(t) + releaseObserver := make(chan struct{}) + backend := dockerWorkloadPTYBackendV1{ + run: func(spec CommandSpec, options RunOptions) error { + writeDockerWorkloadCreateIDV1(spec, options, plan) + return nil + }, + attach: func(context.Context, CommandSpec, string, time.Duration) (dockerPTYAttachmentV1, error) { + return &fakeDockerPTYAttachmentV1{}, nil + }, + resize: func(context.Context, CommandSpec, string, uint32, uint32, time.Duration) error { + return errors.New("resize unavailable") + }, + observe: func(context.Context, CommandSpec, string) (int, error) { + <-releaseObserver + return 143, nil + }, + } + workload, err := prepareDockerWorkloadPTYV1(t.Context(), plan, backend) + if err != nil { + t.Fatal(err) + } + if _, err := workload.Output(); err != nil { + t.Fatal(err) + } + if err := workload.Start(t.Context()); err == nil || !strings.Contains(err.Error(), "resize unavailable") { + t.Fatalf("initial resize error = %v", err) + } + if !workload.Started() { + t.Fatal("Docker-started workload was hidden by the later resize failure") + } + close(releaseObserver) } func TestDockerWorkloadPTYV1ReportsObservationLossAndCallerWaitCancellation(t *testing.T) { plan := controlledSessionWorkloadPlanFixtureV1(t) release := make(chan struct{}) backend := dockerWorkloadPTYBackendV1{ - run: func(CommandSpec, RunOptions) error { return nil }, + run: func(spec CommandSpec, options RunOptions) error { + writeDockerWorkloadCreateIDV1(spec, options, plan) + return nil + }, attach: func(context.Context, CommandSpec, string, time.Duration) (dockerPTYAttachmentV1, error) { return &fakeDockerPTYAttachmentV1{}, nil }, @@ -265,8 +326,9 @@ func TestDockerWorkloadPTYV1FreezesCallerOwnedPlanSlices(t *testing.T) { runs := []CommandSpec{} exit := make(chan int, 1) workload, err := prepareDockerWorkloadPTYV1(t.Context(), plan, dockerWorkloadPTYBackendV1{ - run: func(spec CommandSpec, _ RunOptions) error { + run: func(spec CommandSpec, options RunOptions) error { runs = append(runs, spec) + writeDockerWorkloadCreateIDV1(spec, options, plan) return nil }, attach: func(context.Context, CommandSpec, string, time.Duration) (dockerPTYAttachmentV1, error) { @@ -289,8 +351,14 @@ func TestDockerWorkloadPTYV1FreezesCallerOwnedPlanSlices(t *testing.T) { if _, err := workload.Wait(t.Context()); err != nil { t.Fatal(err) } - if len(runs) != 2 || !reflect.DeepEqual(runs[1].Args, wantStart) { - t.Fatalf("frozen start command = %#v, want %#v", runs, wantStart) + if len(runs) != 2 || !reflect.DeepEqual(runs[1].Args, []string{"start", dockerWorkloadTestContainerIDV1}) { + t.Fatalf("frozen start command = %#v, original planned start %#v", runs, wantStart) + } +} + +func writeDockerWorkloadCreateIDV1(spec CommandSpec, options RunOptions, plan ControlledSessionContainerPlanV1) { + if reflect.DeepEqual(spec.Args, plan.Create.Args) && options.Stdout != nil { + _, _ = fmt.Fprintln(options.Stdout, dockerWorkloadTestContainerIDV1) } } diff --git a/internal/dockerdeploy/testdata/session_channel_helper/main.go b/internal/dockerdeploy/testdata/session_channel_helper/main.go index c922e65..d808900 100644 --- a/internal/dockerdeploy/testdata/session_channel_helper/main.go +++ b/internal/dockerdeploy/testdata/session_channel_helper/main.go @@ -1,6 +1,7 @@ package main import ( + "bytes" "fmt" "os" "os/signal" @@ -12,7 +13,8 @@ import ( func main() { waitForSignal := len(os.Args) == 2 && os.Args[1] == "wait-signal" - if len(os.Args) > 1 && !waitForSignal { + supervise := len(os.Args) == 2 && os.Args[1] == "supervise" + if len(os.Args) > 1 && !waitForSignal && !supervise { fail("unsupported mode %q", os.Args[1]) } var signals chan os.Signal @@ -42,6 +44,10 @@ func main() { _ = os.Remove(probe) fail("private channel mount is writable") } + if supervise { + runSupervisorProof(connection) + return + } if err := controlledsession.WriteRequestV1(connection, controlledsession.RequestV1{Kind: controlledsession.RequestCompleteV1}); err != nil { fail("write complete request: %v", err) } @@ -52,6 +58,86 @@ func main() { } } +func runSupervisorProof(connection readWriteCloser) { + writeRequest(connection, controlledsession.RequestV1{ + Kind: controlledsession.RequestInputV1, Bytes: []byte("stty size; printf 'SIZE-1-DONE\\n'\n"), + }) + var output []byte + resized := false + interruptStarted := false + interruptSent := false + workloadExited := false + for { + event, err := controlledsession.ReadEventV1(connection) + if err != nil { + fail("read session event: %v", err) + } + switch event.Kind { + case controlledsession.EventOutputV1: + output = append(output, event.Bytes...) + if !resized && bytes.Contains(output, []byte("24 80")) && bytes.Contains(output, []byte("SIZE-1-DONE")) { + resized = true + writeRequest(connection, controlledsession.RequestV1{Kind: controlledsession.RequestResizeV1, Columns: 132, Rows: 43}) + writeRequest(connection, controlledsession.RequestV1{ + Kind: controlledsession.RequestInputV1, Bytes: []byte("stty size; printf '\\001\\002\\177\\377'; printf 'SIZE-2-DONE\\n'\n"), + }) + } + if resized && !interruptStarted && bytes.Contains(output, []byte("43 132")) && + bytes.Contains(output, []byte{0x01, 0x02, 0x7f, 0xff}) && bytes.Contains(output, []byte("SIZE-2-DONE")) { + interruptStarted = true + writeRequest(connection, controlledsession.RequestV1{ + Kind: controlledsession.RequestInputV1, Bytes: []byte("printf '\\036SLEEP-ACTIVE\\n'; sleep 30\n"), + }) + } + if interruptStarted && !interruptSent && bytes.Contains(output, append([]byte{0x1e}, []byte("SLEEP-ACTIVE")...)) { + interruptSent = true + writeRequest(connection, controlledsession.RequestV1{Kind: controlledsession.RequestInputV1, Bytes: []byte{0x03}}) + writeRequest(connection, controlledsession.RequestV1{ + Kind: controlledsession.RequestInputV1, Bytes: []byte("printf 'INTERRUPT-DONE\\n'; exit 42\n"), + }) + } + case controlledsession.EventWorkloadExitV1: + if event.WorkloadExit.Status.Code == nil || *event.WorkloadExit.Status.Code != 42 { + fail("unexpected workload exit: %#v, output %q", event.WorkloadExit.Status, output) + } + workloadExited = true + case controlledsession.EventTerminatingV1: + if event.Terminating.Cause != controlledsession.CauseWorkloadExitV1 { + fail("unexpected termination cause %q", event.Terminating.Cause) + } + case controlledsession.EventWorkloadOutputsFinalizedV1: + if event.WorkloadOutputsFinalized.Status != controlledsession.WorkloadOutputFinalizationDrainedV1 || + !workloadExited || !bytes.Contains(output, []byte("INTERRUPT-DONE")) { + fail("unexpected output finalization: %#v, workload exited %t, output %q", event.WorkloadOutputsFinalized, workloadExited, output) + } + writeRequest(connection, controlledsession.RequestV1{Kind: controlledsession.RequestCompleteV1}) + case controlledsession.EventTerminatedV1: + if event.Terminated.Cause != controlledsession.CauseWorkloadExitV1 || + event.Terminated.ControllerFinalizationStatus.Kind != controlledsession.ControllerFinalizationCompletedV1 || + event.Terminated.CleanupStatus.Kind != controlledsession.CleanupStatusSucceededV1 { + fail("unexpected terminal result: %#v", event.Terminated) + } + writeRequest(connection, controlledsession.RequestV1{Kind: controlledsession.RequestAcknowledgeTerminatedV1}) + fmt.Println("PASS: controlled-session supervisor lifecycle") + return + default: + fail("unexpected event kind %q", event.Kind) + } + } +} + +type readWriteCloser interface { + Read([]byte) (int, error) + Write([]byte) (int, error) + Close() error +} + +func writeRequest(connection readWriteCloser, request controlledsession.RequestV1) { + if err := controlledsession.WriteRequestV1(connection, request); err != nil { + fail("write %s request: %v", request.Kind, err) + } +} + func fail(format string, arguments ...any) { _, _ = fmt.Fprintf(os.Stderr, format+"\n", arguments...) os.Exit(1)