diff --git a/.changes/unreleased/+controlled-session-cleanup-containment-loss.yaml b/.changes/unreleased/+controlled-session-cleanup-containment-loss.yaml new file mode 100644 index 00000000..4fdb20d1 --- /dev/null +++ b/.changes/unreleased/+controlled-session-cleanup-containment-loss.yaml @@ -0,0 +1,2 @@ +kind: Changed +body: Report premature controlled-session watchdog exit as cleanup-containment loss, fail the session closed without misreporting Docker observation, and pin watchdog cleanup to the session's verified Docker endpoint. diff --git a/docs/CONTROLLED_SESSION_DESIGN.md b/docs/CONTROLLED_SESSION_DESIGN.md index 480d7739..c823d9a0 100644 --- a/docs/CONTROLLED_SESSION_DESIGN.md +++ b/docs/CONTROLLED_SESSION_DESIGN.md @@ -56,14 +56,16 @@ summary: Capability-scoped execution sessions that inherit Reploy's global conta private channel. A workload that starts before a later startup step fails is still terminated and its output is finalized through the same barrier. Before creating any session resource, the planned controller, workload, and - private-channel ownership plus the session, lease, and boot identities are - now durably recorded in the existing live-run state. Reploy monotonically + private-channel ownership plus the session, lease, boot, and exact verified + local Docker endpoint identities are now durably recorded in the existing + live-run state. Reploy monotonically fills each exact full container ID after Docker creates it, and both IDs are durable before either process starts. Verified cleanup removes that record; failed or unverifiable partial-preparation cleanup retains it. Reploy now derives the immutable watchdog cleanup manifest from the complete recorded ownership before startup. It names only the exact containers and private - channel, carries the host boot identity, + channel, carries the host boot identity and the same pinned local Docker + endpoint used to create both containers, represents the currently absent lease networks and volumes as empty arrays, and omits protocol authority. The watchdog is now launched from the same Reploy executable after that manifest is frozen and before either container @@ -78,15 +80,20 @@ summary: Capability-scoped execution sessions that inherit Reploy's global conta verifies the same resources once more before it exits, and the parent waits boundedly for that exit. The child has no listener and accepts no later resource selection. Because parent and child are the same executable, this - internal manifest has no independent schema-version marker. Next-operation + internal manifest has no independent schema-version marker. A ready + watchdog's premature exit is observed by the attached supervisor. Before + activation it fails startup; after activation it latches the distinct + `cleanup-containment-lost` cause, emits only a safe structured diagnostic + while the controller transport remains usable, and starts ordinary bounded + teardown without misreporting Docker observation as lost. Next-operation restart reconciliation is also implemented: after a prior host boot or an abandoned current-boot owner lease, Reploy discovers any container whose full ID was not recorded by its frozen name, verifies every exact ownership label, removes and verifies both containers and the private channel under one bounded cleanup attempt, and retains incomplete ownership for a later retry. - A watchdog-owned retry loop while Docker remains unavailable is still a - later ownership phase, and controlled-session networking remains a later - phase. + A watchdog-owned retry loop while Docker remains unavailable and a bounded + durable post-crash diagnostic receipt are still later ownership phases, and + 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 @@ -1095,13 +1102,15 @@ preparing -> active -> terminating -> terminated The first accepted termination cause is latched and never rewritten. Causes include controller-requested termination, workload exit, host cancellation, -controller loss, Docker-observation loss, and startup failure. Later events -remain diagnostic observations. Workload status, workload-output-finalization -status, runtime-observation status, controller finalization status, and -pre-delivery cleanup success are reported separately in the session result, so -a late observation or cleanup failure can fail the operation without hiding -its original cause. Controller exit and delivery-tail cleanup are reported -separately by the invoking host operation after teardown. +controller loss, Docker-observation loss, cleanup-containment loss, and startup +failure. Cleanup-containment loss means the ready session watchdog exited +before verified disarm; it does not claim Docker observation was lost. Later +events remain diagnostic observations. Workload status, +workload-output-finalization status, runtime-observation status, controller +finalization status, and pre-delivery cleanup success are reported separately +in the session result, so a late observation or cleanup failure can fail the +operation without hiding its original cause. Controller exit and delivery-tail +cleanup are reported separately by the invoking host operation after teardown. Channel closure is never successful completion. A controller granted the `complete` operation must explicitly send `complete` after receiving @@ -1166,7 +1175,17 @@ Host Reploy starts one short-lived watchdog for each live controlled session. It first creates inert Docker resources and durably records their exact identities. Before starting either container, it passes the watchdog an immutable cleanup manifest containing the exact lease, container, network, -volume, and host boot identities. The attached operation retains one +volume, host boot, and verified local Docker endpoint identities. Controller +and workload creation, attachment, start, observation, resize, stop, and +cleanup use that one pinned endpoint; the watchdog and next-operation +reconciliation reuse it instead of resolving the mutable active Docker context. +Legacy `live-run-queue-v1` ownership written before endpoint pinning remains +readable so abandoned resources do not become unrecoverable after upgrade. +Because that state cannot identify the historical daemon, each legacy cleanup +attempt resolves one currently selected verified local endpoint and pins every +inspection and removal in that attempt to it; exact ownership labels still +gate removal. Newly recorded controlled sessions always require the durable +endpoint. The attached operation retains one end of a private parent pipe. A crash during inert resource creation leaves no untrusted code running and is handled by ordinary next-operation reconciliation. @@ -1179,6 +1198,15 @@ later resource selection, and exits after verified cleanup. Although its underlying Docker connection has ordinary trusted-host authority, its code path is limited to the immutable resource set. +The attached supervisor also observes the watchdog process itself. If a ready +watchdog exits before workload activation, startup fails before workload code +runs. If it exits while the workload is active, Host Reploy latches +`cleanup-containment-lost`, sends a bounded structured diagnostic while the +controller transport remains healthy, and begins normal teardown. The original +termination cause still wins if termination had already begun. Verified host +cleanup may complete the durable ownership record even though containment loss +makes the session result unsuccessful. + If Docker is unavailable, the watchdog retries until Docker returns or the host reboots. If both the attached operation and watchdog are killed, durable labels and deployment-scoped live-run state let the next locked Reploy operation @@ -1247,6 +1275,16 @@ Diagnostics identify which operation failed, what Reploy attempted, whether the session channel or Docker lifecycle was observed, what cleanup ran, and the safe next action. +The target crash-containment work adds a bounded durable incident receipt for +failures that outlive the attached Host Reploy process. It records only +allowlisted lifecycle, observation, exit-status, cleanup, and recovery facts. +It does not duplicate PTY output, environment names or values, secrets, +arbitrary container logs, or unrestricted Docker output. Host Reploy creates +the exact private receipt target before startup and gives the watchdog only the +narrow write authority needed for that target; the child does not select an +arbitrary state path. A parent-liveness EOF proves parent loss but cannot by +itself distinguish `SIGKILL`, an OOM kill, or another abrupt process death. + ## Resource and Timeout Policy Controlled sessions have explicit limits for: diff --git a/internal/controlledsession/lifecycle.go b/internal/controlledsession/lifecycle.go index d04953b4..80667b85 100644 --- a/internal/controlledsession/lifecycle.go +++ b/internal/controlledsession/lifecycle.go @@ -31,6 +31,7 @@ const ( ObservationHostCancelV1 ObservationKindV1 = "host-cancel" ObservationControllerLostV1 ObservationKindV1 = "controller-lost" ObservationRuntimeObservationLostV1 ObservationKindV1 = "runtime-observation-lost" + ObservationCleanupContainmentLostV1 ObservationKindV1 = "cleanup-containment-lost" ObservationStartupFailureV1 ObservationKindV1 = "startup-failure" ObservationWorkloadOutputsFinalizedV1 ObservationKindV1 = "workload-outputs-finalized" ObservationWorkloadOutputsPublishedV1 ObservationKindV1 = "workload-outputs-published" @@ -210,6 +211,11 @@ func (machine *MachineV1) Observe(observation ObservationV1) (TransitionV1, erro // surfaces. Keep that barrier pending until the supervisor explicitly // reports failed closure or its bounded finalization deadline expires. machine.finalizePreActivationOutputsForRuntimeObservationLossLocked(observation.Reason) + case ObservationCleanupContainmentLostV1: + if err := validateCauseObservationV1(observation); err != nil { + return transition, err + } + machine.latchLocked(CauseCleanupContainmentLostV1, &transition) case ObservationStartupFailureV1: if observation.WorkloadStatus != nil || observation.Finish != nil || (observation.WorkloadOutputPending && observation.WorkloadOutputFinalizationStatus != nil) { diff --git a/internal/controlledsession/lifecycle_test.go b/internal/controlledsession/lifecycle_test.go index 62337503..a8f5b0c5 100644 --- a/internal/controlledsession/lifecycle_test.go +++ b/internal/controlledsession/lifecycle_test.go @@ -537,6 +537,16 @@ func TestLifecycleOutputFinalizationExpiryAcceptsLateWorkloadExit(t *testing.T) return err }, }, + { + name: "cleanup containment lost", + cause: CauseCleanupContainmentLostV1, + start: func(machine *MachineV1) error { + _, err := machine.Observe(ObservationV1{ + Kind: ObservationCleanupContainmentLostV1, Reason: "watchdog exited", + }) + return err + }, + }, } for _, test := range tests { @@ -706,6 +716,7 @@ func TestLifecycleFirstAcceptedCauseWinsConcurrentRace(t *testing.T) { {Kind: ObservationHostCancelV1, Reason: "host interrupted"}, {Kind: ObservationControllerLostV1, Reason: "channel closed"}, {Kind: ObservationRuntimeObservationLostV1, Reason: "docker unavailable"}, + {Kind: ObservationCleanupContainmentLostV1, Reason: "watchdog exited"}, {Kind: ObservationWorkloadExitV1, WorkloadStatus: &ProcessStatusV1{Kind: ProcessStatusExitedV1, Code: &code}}, } var wait sync.WaitGroup diff --git a/internal/controlledsession/model.go b/internal/controlledsession/model.go index b91e054b..7ed0cda7 100644 --- a/internal/controlledsession/model.go +++ b/internal/controlledsession/model.go @@ -10,6 +10,7 @@ const ( CauseHostCancelV1 TerminationCauseV1 = "host-cancel" CauseControllerLostV1 TerminationCauseV1 = "controller-lost" CauseRuntimeObservationLostV1 TerminationCauseV1 = "runtime-observation-lost" + CauseCleanupContainmentLostV1 TerminationCauseV1 = "cleanup-containment-lost" CauseStartupFailureV1 TerminationCauseV1 = "startup-failure" ) @@ -242,7 +243,8 @@ func validateRequiredSafeTextV1(field string, value string) error { func validTerminationCauseV1(cause TerminationCauseV1) bool { switch cause { case CauseControllerTerminateV1, CauseWorkloadExitV1, CauseHostCancelV1, - CauseControllerLostV1, CauseRuntimeObservationLostV1, CauseStartupFailureV1: + CauseControllerLostV1, CauseRuntimeObservationLostV1, + CauseCleanupContainmentLostV1, CauseStartupFailureV1: return true default: return false diff --git a/internal/deploy/controlled_session_cleanup_manifest.go b/internal/deploy/controlled_session_cleanup_manifest.go index 64c0ab1c..b2e774f1 100644 --- a/internal/deploy/controlled_session_cleanup_manifest.go +++ b/internal/deploy/controlled_session_cleanup_manifest.go @@ -20,6 +20,7 @@ import ( type ControlledSessionCleanupManifest struct { LiveRunID string `json:"live_run_id"` BootSession string `json:"boot_session"` + DockerEndpoint string `json:"docker_endpoint"` ChannelDirectory string `json:"channel_directory"` Controller ControlledSessionContainerOwnershipV1 `json:"controller"` Workload ControlledSessionContainerOwnershipV1 `json:"workload"` @@ -31,11 +32,12 @@ type ControlledSessionCleanupManifest struct { // from the exact durable ownership record. The session handle is deliberately // omitted because cleanup does not need protocol authority. func ControlledSessionCleanupManifestFromOwnership(ownership ControlledSessionOwnershipV1) (ControlledSessionCleanupManifest, error) { - if err := validateControlledSessionOwnershipV1(ownership); err != nil { + if err := validateCurrentControlledSessionOwnershipV1(ownership); err != nil { return ControlledSessionCleanupManifest{}, fmt.Errorf("controlled-session cleanup manifest ownership: %w", err) } manifest := ControlledSessionCleanupManifest{ LiveRunID: ownership.LiveRunID, BootSession: ownership.BootSession, + DockerEndpoint: ownership.DockerEndpoint, ChannelDirectory: ownership.ChannelDirectory, Controller: ownership.Controller, Workload: ownership.Workload, Networks: []string{}, Volumes: []string{}, @@ -53,6 +55,9 @@ func ValidateControlledSessionCleanupManifest(manifest ControlledSessionCleanupM if err := validateBootSessionIDV1(manifest.BootSession); err != nil { return fmt.Errorf("controlled-session cleanup manifest: %w", err) } + if err := validateControlledSessionDockerEndpointV1(manifest.DockerEndpoint); err != nil { + return fmt.Errorf("controlled-session cleanup manifest: %w", err) + } if !filepath.IsAbs(manifest.ChannelDirectory) || filepath.Clean(manifest.ChannelDirectory) != manifest.ChannelDirectory || !safeRecoveryIdentity(manifest.ChannelDirectory) { return fmt.Errorf("controlled-session cleanup manifest channel directory must be a clean absolute path") } diff --git a/internal/deploy/controlled_session_cleanup_manifest_test.go b/internal/deploy/controlled_session_cleanup_manifest_test.go index 6126715d..03ec6ec5 100644 --- a/internal/deploy/controlled_session_cleanup_manifest_test.go +++ b/internal/deploy/controlled_session_cleanup_manifest_test.go @@ -16,7 +16,7 @@ func TestControlledSessionCleanupManifestDerivesExactDurableOwnership(t *testing t.Fatal(err) } if manifest.LiveRunID != ownership.LiveRunID || manifest.BootSession != ownership.BootSession || - manifest.ChannelDirectory != ownership.ChannelDirectory || manifest.Controller != ownership.Controller || + manifest.DockerEndpoint != ownership.DockerEndpoint || manifest.ChannelDirectory != ownership.ChannelDirectory || manifest.Controller != ownership.Controller || manifest.Workload != ownership.Workload || len(manifest.Networks) != 0 || len(manifest.Volumes) != 0 { t.Fatalf("cleanup manifest = %#v", manifest) } @@ -33,6 +33,15 @@ func TestControlledSessionCleanupManifestDerivesExactDurableOwnership(t *testing } } +func TestControlledSessionCleanupManifestRejectsRemoteDockerEndpoint(t *testing.T) { + ownership := controlledSessionOwnershipFixtureV1(t.TempDir(), "run-0000000000000001", "reploy/env/workload:g-current") + ownership.BootSession = "boot-session" + ownership.DockerEndpoint = "tcp://builder.example:2376" + if _, err := ControlledSessionCleanupManifestFromOwnership(ownership); err == nil || !strings.Contains(err.Error(), "local unix or npipe") { + t.Fatalf("remote Docker endpoint error = %v", err) + } +} + func TestControlledSessionCleanupManifestRejectsInvalidDurableOwnership(t *testing.T) { ownership := controlledSessionOwnershipFixtureV1(t.TempDir(), "run-0000000000000001", "reploy/env/workload:g-current") ownership.BootSession = "boot-session" diff --git a/internal/deploy/live_run_queue.go b/internal/deploy/live_run_queue.go index 0a6916d8..34d2cfc6 100644 --- a/internal/deploy/live_run_queue.go +++ b/internal/deploy/live_run_queue.go @@ -9,6 +9,7 @@ import ( "io" "path/filepath" "regexp" + "strings" "github.com/omry/reploy/internal/canonical" ) @@ -66,6 +67,7 @@ type ControlledSessionOwnershipV1 struct { LiveRunID string `json:"live_run_id"` BootSession string `json:"boot_session"` SessionHandle string `json:"session_handle"` + DockerEndpoint string `json:"docker_endpoint,omitempty"` ChannelDirectory string `json:"channel_directory"` Controller ControlledSessionContainerOwnershipV1 `json:"controller"` Workload ControlledSessionContainerOwnershipV1 `json:"workload"` @@ -289,6 +291,11 @@ func validateControlledSessionOwnershipV1(ownership ControlledSessionOwnershipV1 if !controlledSessionHandlePatternV1.MatchString(ownership.SessionHandle) { return fmt.Errorf("session handle must use session- followed by 64 lowercase hexadecimal characters") } + if ownership.DockerEndpoint != "" { + if err := validateControlledSessionDockerEndpointV1(ownership.DockerEndpoint); err != nil { + return err + } + } if !filepath.IsAbs(ownership.ChannelDirectory) || filepath.Clean(ownership.ChannelDirectory) != ownership.ChannelDirectory || !safeRecoveryIdentity(ownership.ChannelDirectory) { return fmt.Errorf("channel directory must be a clean absolute path") } @@ -307,6 +314,24 @@ func validateControlledSessionOwnershipV1(ownership ControlledSessionOwnershipV1 return nil } +func validateCurrentControlledSessionOwnershipV1(ownership ControlledSessionOwnershipV1) error { + if ownership.DockerEndpoint == "" { + return fmt.Errorf("Docker endpoint must be recorded for a new controlled session") + } + return validateControlledSessionOwnershipV1(ownership) +} + +func validateControlledSessionDockerEndpointV1(endpoint string) error { + if !safeRecoveryIdentity(endpoint) { + return fmt.Errorf("Docker endpoint must be nonempty safe text") + } + scheme, _, found := strings.Cut(endpoint, ":") + if !found || (strings.ToLower(scheme) != "unix" && strings.ToLower(scheme) != "npipe") { + return fmt.Errorf("Docker endpoint must be a local unix or npipe endpoint") + } + return nil +} + func validateControlledSessionContainerOwnershipV1(ownership ControlledSessionContainerOwnershipV1, role string) error { if err := validateControlledSessionContainerOwnershipStateV1(ownership, role); err != nil { return err diff --git a/internal/deploy/live_run_queue_file.go b/internal/deploy/live_run_queue_file.go index afec8440..72341233 100644 --- a/internal/deploy/live_run_queue_file.go +++ b/internal/deploy/live_run_queue_file.go @@ -169,7 +169,7 @@ func (lock *OperationLock) RecordControlledSessionOwnershipV1(ownership Controll return ControlledSessionOwnershipV1{}, fmt.Errorf("controlled session workload generation does not match admitted live run %q", ownership.LiveRunID) } ownership.BootSession = admitted.BootSession - if err := validateControlledSessionOwnershipV1(ownership); err != nil { + if err := validateCurrentControlledSessionOwnershipV1(ownership); err != nil { return ControlledSessionOwnershipV1{}, err } insert := sort.Search(len(queue.ControlledSessions), func(index int) bool { @@ -230,7 +230,7 @@ func mergeControlledSessionOwnershipV1( if err != nil { return ControlledSessionOwnershipV1{}, err } - if err := validateControlledSessionOwnershipV1(merged); err != nil { + if err := validateCurrentControlledSessionOwnershipV1(merged); err != nil { return ControlledSessionOwnershipV1{}, err } return merged, nil diff --git a/internal/deploy/live_run_queue_file_test.go b/internal/deploy/live_run_queue_file_test.go index efd00a8f..0257d266 100644 --- a/internal/deploy/live_run_queue_file_test.go +++ b/internal/deploy/live_run_queue_file_test.go @@ -80,6 +80,11 @@ func TestOperationLockRecordsExactControlledSessionOwnership(t *testing.T) { planned := ownership planned.Controller.ID = "" planned.Workload.ID = "" + missingEndpoint := planned + missingEndpoint.DockerEndpoint = "" + if _, err := lock.RecordControlledSessionOwnershipV1(missingEndpoint); err == nil || !strings.Contains(err.Error(), "Docker endpoint must be recorded") { + t.Fatalf("missing Docker endpoint error = %v", err) + } recorded, err := lock.RecordControlledSessionOwnershipV1(planned) if err != nil { t.Fatal(err) @@ -122,6 +127,32 @@ func TestOperationLockRecordsExactControlledSessionOwnership(t *testing.T) { } } +func TestLiveRunQueueV1DecodesLegacyControlledSessionWithoutDockerEndpoint(t *testing.T) { + dir := t.TempDir() + bootSession, err := CurrentBootSessionIDV1() + if err != nil { + t.Fatal(err) + } + ownership := controlledSessionOwnershipFixtureV1( + dir, "run-0000000000000001", "reploy/env/workload:g-current", + ) + ownership.BootSession = bootSession + ownership.DockerEndpoint = "" + queue := NewLiveRunQueueV1() + queue.ControlledSessions = []ControlledSessionOwnershipV1{ownership} + content, err := EncodeLiveRunQueueV1(queue) + if err != nil { + t.Fatal(err) + } + if bytes.Contains(content, []byte("docker_endpoint")) { + t.Fatalf("legacy queue unexpectedly gained a Docker endpoint: %s", content) + } + decoded, err := DecodeLiveRunQueueV1(content) + if err != nil || !reflect.DeepEqual(decoded, queue) { + t.Fatalf("legacy queue = %#v, error=%v", decoded, err) + } +} + func TestOperationLockControlledSessionOwnershipWriteFailurePreservesQueue(t *testing.T) { dir := t.TempDir() lock, err := AcquireOperationLock(t.Context(), dir) @@ -208,6 +239,7 @@ func controlledSessionOwnershipFixtureV1(dir string, runID string, generation st } return ControlledSessionOwnershipV1{ LiveRunID: runID, SessionHandle: "session-" + strings.Repeat("a", 64), + DockerEndpoint: "unix:///var/run/docker.sock", ChannelDirectory: filepath.Join(dir, ".reploy", "sessions", runID), Controller: container("controller", strings.Repeat("a", 64), "controller", "reploy/env/controller:g-current", "1"), Workload: container("workload", strings.Repeat("b", 64), "workload", generation, "2"), diff --git a/internal/dockerdeploy/compose.go b/internal/dockerdeploy/compose.go index 0312d103..eaf5bf88 100644 --- a/internal/dockerdeploy/compose.go +++ b/internal/dockerdeploy/compose.go @@ -89,6 +89,18 @@ func bindPinnedDockerCommandRunnerV1( return pinned, run, nil } +func commandRunnerForPinnedDockerEndpointV1(endpoint string, run commandRunner) (commandRunner, error) { + if run == nil { + return nil, fmt.Errorf("pin Docker endpoint requires a command runner") + } + if !localDockerEndpointV1(endpoint) { + return nil, fmt.Errorf("controlled-session Docker endpoint %q is not local", endpoint) + } + return func(spec CommandSpec, options RunOptions) error { + return run(pinDockerEndpointV1(spec, endpoint), options) + }, nil +} + // runCommandWithoutDockerPreflight executes non-Docker commands and Docker // commands whose exact local endpoint was already pinned by runDockerCommand. // Recognizable unpinned Docker commands fail closed. diff --git a/internal/dockerdeploy/compose_test.go b/internal/dockerdeploy/compose_test.go index 6c95af6a..e8dd2b01 100644 --- a/internal/dockerdeploy/compose_test.go +++ b/internal/dockerdeploy/compose_test.go @@ -110,6 +110,34 @@ func TestRunCommandPassesDockerPreflightTimeout(t *testing.T) { } } +func TestCommandRunnerForPinnedDockerEndpointV1PinsEveryCommand(t *testing.T) { + const endpoint = "unix:///session-engine.sock" + var commands []CommandSpec + run, err := commandRunnerForPinnedDockerEndpointV1(endpoint, func(spec CommandSpec, _ RunOptions) error { + commands = append(commands, spec) + return nil + }) + if err != nil { + t.Fatal(err) + } + for _, args := range [][]string{{"container", "inspect", "id"}, {"container", "rm", "--force", "id"}} { + if err := run(CommandSpec{Name: "docker", Args: args}, RunOptions{}); err != nil { + t.Fatal(err) + } + } + for _, command := range commands { + if got := commandEnvironmentValueV1(command, "DOCKER_HOST"); got != endpoint { + t.Fatalf("pinned Docker host = %q", got) + } + if contextName, found := commandSpecEnvironmentValueV1(command, "DOCKER_CONTEXT"); !found || contextName != "" { + t.Fatalf("pinned Docker context = %q, found=%t", contextName, found) + } + } + if _, err := commandRunnerForPinnedDockerEndpointV1("tcp://remote.example:2376", runCommandWithoutDockerPreflight); err == nil { + t.Fatal("remote endpoint was accepted") + } +} + func TestRunCommandWithoutDockerPreflightRunsKnownFollowup(t *testing.T) { dir := t.TempDir() writeFakeCommand( diff --git a/internal/dockerdeploy/control_admission_modes_test.go b/internal/dockerdeploy/control_admission_modes_test.go index bdc54ed7..8200daf5 100644 --- a/internal/dockerdeploy/control_admission_modes_test.go +++ b/internal/dockerdeploy/control_admission_modes_test.go @@ -162,7 +162,7 @@ func TestAdmitControlOperationV1ForceStopsControlledSessionContainersAndRetainsO t.Fatal(err) } ownership, err := operation.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFromPlanV1( - plan, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1, + plan, controlledSessionTestDockerEndpointV1, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1, )) if err != nil { t.Fatal(err) @@ -218,7 +218,7 @@ func TestAdmitControlOperationV1ForcePreservesControlledSessionOnPartialStopFail t.Fatal(err) } ownership, err := operation.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFromPlanV1( - plan, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1, + plan, controlledSessionTestDockerEndpointV1, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1, )) if err != nil { t.Fatal(err) diff --git a/internal/dockerdeploy/controlled_session_controller_test.go b/internal/dockerdeploy/controlled_session_controller_test.go index 2e1d661b..5954a38d 100644 --- a/internal/dockerdeploy/controlled_session_controller_test.go +++ b/internal/dockerdeploy/controlled_session_controller_test.go @@ -12,7 +12,10 @@ import ( "github.com/omry/reploy/internal/controlledsession" ) -const dockerControllerTestContainerIDV1 = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" +const ( + dockerControllerTestContainerIDV1 = "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" + controlledSessionTestDockerEndpointV1 = "unix:///var/run/docker.sock" +) func TestDockerControllerV1OrdersChannelCreateStartAndExactLifecycle(t *testing.T) { plan := controlledSessionControllerPlanFixtureV1(t) diff --git a/internal/dockerdeploy/controlled_session_supervisor.go b/internal/dockerdeploy/controlled_session_supervisor.go index 32657735..77a8d826 100644 --- a/internal/dockerdeploy/controlled_session_supervisor.go +++ b/internal/dockerdeploy/controlled_session_supervisor.go @@ -128,6 +128,9 @@ type controlledSessionSupervisorV1 struct { controllerOwnershipIncomplete bool cleanupManifest deploy.ControlledSessionCleanupManifest watchdog controlledSessionWatchdogRuntimeV1 + watchdogExitObserved bool + watchdogExitAfterTermination bool + watchdogExitErr error preCleanupSucceeded bool transportHealthy bool @@ -169,12 +172,16 @@ func RunControlledSessionV1( if err := operation.RequireQueueEntryLeaseHeldV1(plan.LiveRunID); err != nil { return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionV1(operation, plan.LiveRunID, fmt.Errorf("controlled-session admission ownership: %w", err)) } + dockerEndpoint, bindSessionDocker, err := bindControlledSessionDockerEndpointV1(ctx, controlledSessionCommandSpecV1(plan.Controller.Create)) + if err != nil { + return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionV1(operation, plan.LiveRunID, fmt.Errorf("bind controlled-session Docker endpoint: %w", err)) + } operationReleaseAttempted := false ownershipRecorded := false partialPreparationCleanupVerified := false controllerID := "" persistOwnership := func(controllerID string, workloadID string) (deploy.ControlledSessionOwnershipV1, error) { - ownership := controlledSessionOwnershipFromPlanV1(plan, controllerID, workloadID) + ownership := controlledSessionOwnershipFromPlanV1(plan, dockerEndpoint, controllerID, workloadID) recorded, err := operation.RecordControlledSessionOwnershipV1(ownership) if err != nil { return deploy.ControlledSessionOwnershipV1{}, fmt.Errorf("persist controlled-session ownership: %w", err) @@ -192,16 +199,22 @@ func RunControlledSessionV1( return &privateControlledSessionChannelRuntimeV1{channel: channel}, nil }, prepareController: func(ctx context.Context, plan ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) { - return prepareDockerControllerWithCleanupVerificationV1(ctx, plan, func() { - partialPreparationCleanupVerified = true + return prepareDockerControllerV1(ctx, plan, dockerControllerBackendV1{ + bind: bindSessionDocker, observe: observeDockerContainerExitV1, + requireReadyChannel: requirePreparedControlledSessionControllerChannelV1, + recordNoContainer: func() { partialPreparationCleanupVerified = true }, }) }, prepareWorkload: func(ctx context.Context, plan ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { - return prepareDockerWorkloadPTYWithContainerIDV1(ctx, plan, func(workloadID string) error { - _, err := persistOwnership(controllerID, workloadID) - return err - }, func() { - partialPreparationCleanupVerified = true + return prepareDockerWorkloadPTYV1(ctx, plan, dockerWorkloadPTYBackendV1{ + bind: bindSessionDocker, + recordContainerID: func(workloadID string) error { + _, err := persistOwnership(controllerID, workloadID) + return err + }, + recordRollbackVerified: func() { partialPreparationCleanupVerified = true }, + attach: attachDockerContainerPTYV1, resize: resizeDockerContainerPTYV1, + observe: observeDockerContainerExitV1, }) }, recordPlannedOwnership: func() error { @@ -241,6 +254,24 @@ func RunControlledSessionV1( return result, errors.Join(runErr, completionErr) } +func bindControlledSessionDockerEndpointV1( + ctx context.Context, + docker CommandSpec, +) (string, func(context.Context, CommandSpec, time.Duration) (CommandSpec, commandRunner, error), error) { + pinnedDocker, run, err := bindPinnedDockerCommandRunnerV1(ctx, docker, defaultDockerPreflightTimeout) + if err != nil { + return "", nil, err + } + endpoint := commandEnvironmentValueV1(pinnedDocker, "DOCKER_HOST") + bind := func(_ context.Context, spec CommandSpec, _ time.Duration) (CommandSpec, commandRunner, error) { + if spec.Name != pinnedDocker.Name { + return CommandSpec{}, nil, fmt.Errorf("controlled-session Docker executable changed from %q to %q", pinnedDocker.Name, spec.Name) + } + return pinDockerEndpointV1(spec, endpoint), run, nil + } + return endpoint, bind, nil +} + func controlledSessionChannelAbsentV1(path string) bool { _, err := os.Lstat(path) return errors.Is(err, os.ErrNotExist) @@ -250,7 +281,7 @@ func controlledSessionPreparationCanCompleteV1(cleaned bool, ownershipRecorded b return cleaned && (!ownershipRecorded || operationReleaseAttempted || partialCleanupVerified) } -func controlledSessionOwnershipFromPlanV1(plan ControlledSessionExecutionPlanV1, controllerID string, workloadID string) deploy.ControlledSessionOwnershipV1 { +func controlledSessionOwnershipFromPlanV1(plan ControlledSessionExecutionPlanV1, dockerEndpoint string, controllerID string, workloadID string) deploy.ControlledSessionOwnershipV1 { container := func(plan ControlledSessionContainerPlanV1, id string) deploy.ControlledSessionContainerOwnershipV1 { return deploy.ControlledSessionContainerOwnershipV1{ Role: string(plan.Role), ID: id, Name: plan.Container, DeploymentID: plan.DeploymentID, @@ -259,6 +290,7 @@ func controlledSessionOwnershipFromPlanV1(plan ControlledSessionExecutionPlanV1, } return deploy.ControlledSessionOwnershipV1{ LiveRunID: plan.LiveRunID, SessionHandle: plan.Authorization.Handle, + DockerEndpoint: dockerEndpoint, ChannelDirectory: plan.Channel.HostDirectory, Controller: container(plan.Controller, controllerID), Workload: container(plan.Workload, workloadID), } @@ -390,8 +422,10 @@ func (supervisor *controlledSessionSupervisorV1) run(ctx context.Context) (Contr supervisor.stopAndObserveWorkload() supervisor.finalizeWorkloadOutput() supervisor.waitForControllerFinalization() + supervisor.observeWatchdogExitIfDone() preDeliveryCleanup, preDeliveryRecovery, cleanupErr := supervisor.cleanupWorkload() + supervisor.observeWatchdogExitIfDone() supervisor.preCleanupSucceeded = preDeliveryCleanup.Kind == controlledsession.CleanupStatusSucceededV1 finish := supervisor.finishStatus(preDeliveryCleanup, preDeliveryRecovery) transition, finishErr := supervisor.observe(controlledsession.ObservationV1{Kind: controlledsession.ObservationFinishedV1, Finish: &finish}) @@ -474,14 +508,30 @@ func (supervisor *controlledSessionSupervisorV1) prepare(ctx context.Context) er return fmt.Errorf("launch controlled-session watchdog: %w", err) } supervisor.watchdog = watchdog + watchdogCtx, cancelWatchdogCtx := context.WithCancelCause(ctx) + defer cancelWatchdogCtx(nil) + go func() { + select { + case <-watchdog.Done(): + cancelWatchdogCtx(unexpectedControlledSessionWatchdogExitV1(watchdog)) + case <-watchdogCtx.Done(): + } + }() + ctx = watchdogCtx } if err := controller.Start(ctx); err != nil { + if watchdogErr := supervisor.currentWatchdogExitError(); watchdogErr != nil { + return watchdogErr + } 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 { + if watchdogErr := supervisor.currentWatchdogExitError(); watchdogErr != nil { + return watchdogErr + } return fmt.Errorf("claim controlled-session controller channel: %w", err) } bridge, err := controlledsession.StartSessionIOBridgeV1(transport, output, supervisor.handleRequest) @@ -494,13 +544,39 @@ func (supervisor *controlledSessionSupervisorV1) prepare(ctx context.Context) er if supervisor.workloadStarted { supervisor.workloadResult = observeControlledSessionProcessV1(workload.Wait) } + if watchdogErr := supervisor.currentWatchdogExitError(); watchdogErr != nil { + return watchdogErr + } return fmt.Errorf("start controlled-session workload: %w", err) } supervisor.workloadStarted = true supervisor.workloadResult = observeControlledSessionProcessV1(workload.Wait) + if watchdogErr := supervisor.currentWatchdogExitError(); watchdogErr != nil { + return watchdogErr + } return nil } +func unexpectedControlledSessionWatchdogExitV1(watchdog controlledSessionWatchdogRuntimeV1) error { + if err := watchdog.ExitError(); err != nil { + return fmt.Errorf("controlled-session cleanup watchdog exited unexpectedly: %w", err) + } + return fmt.Errorf("controlled-session cleanup watchdog exited unexpectedly") +} + +func (supervisor *controlledSessionSupervisorV1) currentWatchdogExitError() error { + if supervisor.watchdog == nil { + return nil + } + select { + case <-supervisor.watchdog.Done(): + supervisor.watchdogExitObserved = true + return unexpectedControlledSessionWatchdogExitV1(supervisor.watchdog) + default: + return nil + } +} + func validateControlledSessionCleanupManifestForRuntimeV1( manifest deploy.ControlledSessionCleanupManifest, plan ControlledSessionExecutionPlanV1, @@ -510,9 +586,9 @@ func validateControlledSessionCleanupManifestForRuntimeV1( if err := deploy.ValidateControlledSessionCleanupManifest(manifest); err != nil { return fmt.Errorf("validate controlled-session cleanup manifest: %w", err) } - expected := controlledSessionOwnershipFromPlanV1(plan, controllerID, workloadID) + expected := controlledSessionOwnershipFromPlanV1(plan, manifest.DockerEndpoint, controllerID, workloadID) if manifest.LiveRunID != expected.LiveRunID || manifest.ChannelDirectory != expected.ChannelDirectory || - manifest.Controller != expected.Controller || manifest.Workload != expected.Workload { + manifest.DockerEndpoint != expected.DockerEndpoint || manifest.Controller != expected.Controller || manifest.Workload != expected.Workload { return fmt.Errorf("controlled-session cleanup manifest does not match the exact prepared resources") } if len(manifest.Networks) != 0 || len(manifest.Volumes) != 0 { @@ -591,6 +667,8 @@ func (supervisor *controlledSessionSupervisorV1) waitForTermination(ctx context. outputDone = nil supervisor.observeOutputTermination() case <-supervisor.stateChanged: + case <-supervisor.watchdogDone(): + supervisor.observeWatchdogExit() case <-ctx.Done(): _, err := supervisor.observe(controlledsession.ObservationV1{ Kind: controlledsession.ObservationHostCancelV1, Reason: "host operation was canceled", @@ -602,6 +680,51 @@ func (supervisor *controlledSessionSupervisorV1) waitForTermination(ctx context. } } +func (supervisor *controlledSessionSupervisorV1) watchdogDone() <-chan struct{} { + if supervisor.watchdog == nil || supervisor.watchdogExitObserved { + return nil + } + return supervisor.watchdog.Done() +} + +func (supervisor *controlledSessionSupervisorV1) observeWatchdogExit() { + if supervisor.watchdog == nil || supervisor.watchdogExitObserved { + return + } + supervisor.watchdogExitObserved = true + detail := unexpectedControlledSessionWatchdogExitV1(supervisor.watchdog) + supervisor.watchdogExitErr = detail + if supervisor.machine.Snapshot().State == controlledsession.StateTerminatedV1 { + supervisor.watchdogExitAfterTermination = true + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, detail) + return + } + _, observeErr := supervisor.observe(controlledsession.ObservationV1{ + Kind: controlledsession.ObservationCleanupContainmentLostV1, + Reason: "controlled-session cleanup containment was lost", + }) + supervisor.diagnosticErr = errors.Join(supervisor.diagnosticErr, detail, observeErr) + if supervisor.transportHealthy && supervisor.bridge != nil { + if err := supervisor.sendPreFinalizationLifecycleEvent(controlledsession.EventV1{ + Kind: controlledsession.EventDiagnosticV1, + Diagnostic: &controlledsession.DiagnosticV1{ + Code: "cleanup_containment_lost", + Message: "controlled-session cleanup containment was lost", + }, + }); err != nil { + supervisor.loseTransport("send cleanup-containment-loss diagnostic", err) + } + } +} + +func (supervisor *controlledSessionSupervisorV1) observeWatchdogExitIfDone() { + select { + case <-supervisor.watchdogDone(): + supervisor.observeWatchdogExit() + default: + } +} + func (supervisor *controlledSessionSupervisorV1) observeOutputTermination() { result, stopped := supervisor.bridge.OutputTerminalResult() if !stopped { @@ -825,6 +948,8 @@ func (supervisor *controlledSessionSupervisorV1) waitForControllerFinalization() supervisor.observeControllerLoss("controller process exited during finalization", result.err) case <-supervisor.bridge.RequestsDone(): supervisor.observeRequestFailure() + case <-supervisor.watchdogDone(): + supervisor.observeWatchdogExit() case <-timer.C: _, err := supervisor.observe(controlledsession.ObservationV1{Kind: controlledsession.ObservationControllerFinalizationExpiredV1}) if err != nil && supervisor.machine.Snapshot().AwaitingControllerFinalization { @@ -885,6 +1010,9 @@ func (supervisor *controlledSessionSupervisorV1) waitForResultAcknowledgement() return supervisor.machine.Snapshot().ResultAcknowledged case result := <-supervisor.controllerResult: supervisor.controllerObserved = &result + case <-supervisor.watchdogDone(): + supervisor.observeWatchdogExit() + return supervisor.machine.Snapshot().ResultAcknowledged case <-timer.C: return supervisor.machine.Snapshot().ResultAcknowledged } @@ -897,6 +1025,7 @@ func (supervisor *controlledSessionSupervisorV1) cleanupDeliveryTail() ( controlledsession.CleanupStatusV1, controlledsession.RecoveryActionV1, ) { + supervisor.observeWatchdogExitIfDone() if supervisor.bridge != nil { supervisor.bridge.StopRequests() } @@ -946,6 +1075,7 @@ func (supervisor *controlledSessionSupervisorV1) cleanupDeliveryTail() ( controllerCleanupErr := supervisor.controller.Cleanup(cleanupCtx) cleanupCancel() cleanupErr = errors.Join(cleanupErr, controllerCleanupErr) + supervisor.observeWatchdogExitIfDone() if controllerCleanupErr == nil && supervisor.controllerOwnershipIncomplete { if supervisor.backend.recordControllerRollback != nil { supervisor.backend.recordControllerRollback() @@ -958,23 +1088,25 @@ func (supervisor *controlledSessionSupervisorV1) cleanupDeliveryTail() ( status = supervisor.controllerObserved.status cleanupErr = errors.Join(cleanupErr, supervisor.controllerObserved.err) } + if supervisor.watchdog != nil && supervisor.preCleanupSucceeded { + supervisor.observeWatchdogExitIfDone() + if supervisor.watchdogExitObserved { + cleanupErr = errors.Join(cleanupErr, supervisor.watchdog.Close()) + } else if cleanupErr == nil { + disarmCtx, disarmCancel := context.WithTimeout(context.Background(), supervisor.options.CleanupTimeout) + cleanupErr = errors.Join(cleanupErr, supervisor.watchdog.Disarm(disarmCtx)) + disarmCancel() + } + } + if supervisor.watchdogExitAfterTermination { + cleanupErr = errors.Join(cleanupErr, supervisor.watchdogExitErr) + } 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 } - if supervisor.watchdog != nil && supervisor.preCleanupSucceeded { - disarmCtx, disarmCancel := context.WithTimeout(context.Background(), supervisor.options.CleanupTimeout) - cleanupErr = errors.Join(cleanupErr, supervisor.watchdog.Disarm(disarmCtx)) - disarmCancel() - 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 } diff --git a/internal/dockerdeploy/controlled_session_supervisor_test.go b/internal/dockerdeploy/controlled_session_supervisor_test.go index 5737f6ce..8dbe8ea2 100644 --- a/internal/dockerdeploy/controlled_session_supervisor_test.go +++ b/internal/dockerdeploy/controlled_session_supervisor_test.go @@ -54,7 +54,7 @@ func TestRunControlledSessionV1OwnsNormalLifecycle(t *testing.T) { recordPlannedOwnership: func() error { return nil }, recordControllerOwnership: func(string) error { return nil }, recordOwnership: func(controllerID string, workloadID string) (deploy.ControlledSessionCleanupManifest, error) { - ownership := controlledSessionOwnershipFromPlanV1(plan, controllerID, workloadID) + ownership := controlledSessionOwnershipFromPlanV1(plan, controlledSessionTestDockerEndpointV1, controllerID, workloadID) ownership.BootSession = "boot-session" return deploy.ControlledSessionCleanupManifestFromOwnership(ownership) }, @@ -120,6 +120,132 @@ func TestRunControlledSessionV1OwnsNormalLifecycle(t *testing.T) { } } +func TestBindControlledSessionDockerEndpointV1SelectsOnceForBothContainers(t *testing.T) { + const endpoint = "unix:///session-engine.sock" + preflights := 0 + restore := stubDockerPreflight(t, func(context.Context, CommandSpec, time.Duration) (string, error) { + preflights++ + return endpoint, nil + }) + defer restore() + + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + selected, bind, err := bindControlledSessionDockerEndpointV1(t.Context(), controlledSessionCommandSpecV1(plan.Controller.Create)) + if err != nil { + t.Fatal(err) + } + if selected != endpoint || preflights != 1 { + t.Fatalf("selected endpoint=%q preflights=%d", selected, preflights) + } + for _, container := range []ControlledSessionContainerPlanV1{plan.Controller, plan.Workload} { + pinned, run, err := bind(t.Context(), controlledSessionCommandSpecV1(container.Create), time.Second) + if err != nil || run == nil { + t.Fatalf("bind %s = %#v, %v", container.Role, pinned, err) + } + if host, found := commandSpecEnvironmentValueV1(pinned, "DOCKER_HOST"); !found || host != endpoint { + t.Fatalf("%s Docker host=%q found=%t", container.Role, host, found) + } + if contextName, found := commandSpecEnvironmentValueV1(pinned, "DOCKER_CONTEXT"); !found || contextName != "" { + t.Fatalf("%s Docker context=%q found=%t", container.Role, contextName, found) + } + } + if preflights != 1 { + t.Fatalf("container binding repeated endpoint preflight %d times", preflights) + } +} + +func TestRunControlledSessionV1FailsClosedAfterWatchdogExit(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + requests := make(chan controlledsession.RequestV1, 8) + requests <- controlledsession.RequestV1{Kind: controlledsession.RequestInputV1, Bytes: []byte("ready")} + 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}, + } + } + } + workload := newFakeControlledSessionWorkloadV1(nil, 143) + workload.exitOnStart = false + watchdog := &fakeControlledSessionWatchdogV1{} + exitErr := errors.New("watchdog process failed") + + type completion struct { + result ControlledSessionRunResultV1 + err error + } + done := make(chan completion, 1) + go func() { + 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 + }, + recordPlannedOwnership: func() error { return nil }, + recordControllerOwnership: func(string) error { return nil }, + recordOwnership: func(controllerID string, workloadID string) (deploy.ControlledSessionCleanupManifest, error) { + ownership := controlledSessionOwnershipFromPlanV1(plan, controlledSessionTestDockerEndpointV1, controllerID, workloadID) + ownership.BootSession = "boot-session" + return deploy.ControlledSessionCleanupManifestFromOwnership(ownership) + }, + startWatchdog: func(context.Context, deploy.ControlledSessionCleanupManifest) (controlledSessionWatchdogRuntimeV1, error) { + return watchdog, nil + }, + now: time.Now, + }) + done <- completion{result: result, err: err} + }() + + select { + case <-workload.inputStarted: + watchdog.exitUnexpectedly(exitErr) + case <-time.After(3 * time.Second): + t.Fatal("controlled session did not become active") + } + + var completed completion + select { + case completed = <-done: + case <-time.After(3 * time.Second): + t.Fatal("controlled session did not stop after watchdog exit") + } + if !errors.Is(completed.err, exitErr) { + t.Fatalf("error = %v", completed.err) + } + if completed.result.SessionResult.Cause != controlledsession.CauseCleanupContainmentLostV1 || + completed.result.SessionResult.RuntimeObservationStatus.Kind != controlledsession.RuntimeObservationMaintainedV1 || + completed.result.SessionResult.CleanupStatus.Kind != controlledsession.CleanupStatusSucceededV1 || + completed.result.DeliveryTailCleanupStatus.Kind != controlledsession.CleanupStatusSucceededV1 { + t.Fatalf("session result = %#v", completed.result) + } + if !workload.gracefulStopped || !workload.cleaned || !controller.cleaned || !watchdog.closed || watchdog.disarmed { + t.Fatalf("cleanup = workload stopped %t/cleaned %t, controller cleaned %t, watchdog closed %t/disarmed %t", + workload.gracefulStopped, workload.cleaned, controller.cleaned, watchdog.closed, watchdog.disarmed) + } + var diagnosticFound bool + for _, event := range transport.snapshotEvents() { + if event.Kind == controlledsession.EventDiagnosticV1 && event.Diagnostic != nil && + event.Diagnostic.Code == "cleanup_containment_lost" { + diagnosticFound = true + } + } + if !diagnosticFound { + t.Fatalf("cleanup-containment-loss diagnostic missing from %#v", transport.snapshotEvents()) + } +} + func TestRunControlledSessionV1PersistsExactOwnershipBeforeStarting(t *testing.T) { plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) controller := newFakeControlledSessionProcessV1() @@ -403,7 +529,7 @@ func TestRunControlledSessionV1RejectsCleanupResourcesNotInDurableOwnership(t *t recordPlannedOwnership: func() error { return nil }, recordControllerOwnership: func(string) error { return nil }, recordOwnership: func(controllerID string, workloadID string) (deploy.ControlledSessionCleanupManifest, error) { - ownership := controlledSessionOwnershipFromPlanV1(plan, controllerID, workloadID) + ownership := controlledSessionOwnershipFromPlanV1(plan, controlledSessionTestDockerEndpointV1, controllerID, workloadID) ownership.BootSession = "boot-session" manifest, manifestErr := deploy.ControlledSessionCleanupManifestFromOwnership(ownership) manifest.Networks = []string{"unrelated-network"} @@ -446,7 +572,7 @@ func TestRunControlledSessionV1DoesNotStartWhenWatchdogLaunchFails(t *testing.T) recordPlannedOwnership: func() error { return nil }, recordControllerOwnership: func(string) error { return nil }, recordOwnership: func(controllerID string, workloadID string) (deploy.ControlledSessionCleanupManifest, error) { - ownership := controlledSessionOwnershipFromPlanV1(plan, controllerID, workloadID) + ownership := controlledSessionOwnershipFromPlanV1(plan, controlledSessionTestDockerEndpointV1, controllerID, workloadID) ownership.BootSession = "boot-session" return deploy.ControlledSessionCleanupManifestFromOwnership(ownership) }, @@ -470,6 +596,67 @@ func TestRunControlledSessionV1DoesNotStartWhenWatchdogLaunchFails(t *testing.T) } } +func TestRunControlledSessionV1FailsStartupWhenReadyWatchdogExits(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + controller := newFakeControlledSessionProcessV1() + workload := newFakeControlledSessionWorkloadV1(nil, 0) + channel := &fakeControlledSessionChannelV1{blockClaim: true, claimStarted: make(chan struct{})} + watchdog := &fakeControlledSessionWatchdogV1{} + exitErr := errors.New("watchdog exited after readiness") + + type completion struct { + result ControlledSessionRunResultV1 + err error + } + done := make(chan completion, 1) + go func() { + 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 + }, + recordPlannedOwnership: func() error { return nil }, + recordControllerOwnership: func(string) error { return nil }, + recordOwnership: func(controllerID string, workloadID string) (deploy.ControlledSessionCleanupManifest, error) { + ownership := controlledSessionOwnershipFromPlanV1(plan, controlledSessionTestDockerEndpointV1, controllerID, workloadID) + ownership.BootSession = "boot-session" + return deploy.ControlledSessionCleanupManifestFromOwnership(ownership) + }, + startWatchdog: func(context.Context, deploy.ControlledSessionCleanupManifest) (controlledSessionWatchdogRuntimeV1, error) { + return watchdog, nil + }, + now: time.Now, + }) + done <- completion{result: result, err: err} + }() + + select { + case <-channel.claimStarted: + watchdog.exitUnexpectedly(exitErr) + case <-time.After(3 * time.Second): + t.Fatal("controller channel claim did not begin") + } + var completed completion + select { + case completed = <-done: + case <-time.After(3 * time.Second): + t.Fatal("startup did not stop after watchdog exit") + } + if !errors.Is(completed.err, exitErr) || + completed.result.SessionResult.Cause != controlledsession.CauseStartupFailureV1 { + t.Fatalf("startup result = %#v, %v", completed.result, completed.err) + } + if workload.started || !workload.cleaned || !controller.cleaned || !channel.closed || !watchdog.closed || watchdog.disarmed { + t.Fatalf("startup cleanup = workload started %t/cleaned %t, controller cleaned %t, channel closed %t, watchdog closed %t/disarmed %t", + workload.started, workload.cleaned, controller.cleaned, channel.closed, watchdog.closed, watchdog.disarmed) + } +} + func TestControlledSessionWatchdogDisarmRequiresVerifiedCompleteCleanup(t *testing.T) { disarmErr := errors.New("injected disarm failure") for _, test := range []struct { @@ -505,6 +692,45 @@ func TestControlledSessionWatchdogDisarmRequiresVerifiedCompleteCleanup(t *testi } } +func TestControlledSessionDeliveryTailObservesWatchdogExitBeforeDisarm(t *testing.T) { + exitErr := errors.New("watchdog process failed") + watchdog := &fakeControlledSessionWatchdogV1{} + controller := newFakeControlledSessionProcessV1() + controller.onCleanup = func() { watchdog.exitUnexpectedly(exitErr) } + machine, err := controlledsession.NewMachineV1(controlledSessionControllerIntegrationPlanV1( + t, "test-image", []string{"/controller"}, + ).Authorization) + if err != nil { + t.Fatal(err) + } + if _, err := machine.Observe(controlledsession.ObservationV1{Kind: controlledsession.ObservationStartupFailureV1, Reason: "test"}); err != nil { + t.Fatal(err) + } + finish := controlledsession.FinishV1{ + WorkloadStatus: controlledsession.ProcessStatusV1{Kind: controlledsession.ProcessStatusUnknownV1}, + WorkloadOutputFinalizationStatus: controlledsession.WorkloadOutputFinalizationStatusV1{Kind: controlledsession.WorkloadOutputFinalizationDrainedV1}, + ControllerFinalizationStatus: controlledsession.ControllerFinalizationStatusV1{Kind: controlledsession.ControllerFinalizationStartupFailedV1, Reason: "test"}, + CleanupStatus: controlledsession.CleanupStatusV1{Kind: controlledsession.CleanupStatusSucceededV1}, + RecoveryAction: controlledsession.RecoveryNoneV1, + } + if _, err := machine.Observe(controlledsession.ObservationV1{Kind: controlledsession.ObservationFinishedV1, Finish: &finish}); err != nil { + t.Fatal(err) + } + supervisor := &controlledSessionSupervisorV1{ + options: testControlledSessionRunOptionsV1(), machine: machine, + channel: &fakeControlledSessionChannelV1{}, controller: controller, + watchdog: watchdog, preCleanupSucceeded: true, + } + _, cleanup, recovery := supervisor.cleanupDeliveryTail() + if cleanup.Kind != controlledsession.CleanupStatusFailedV1 || recovery != controlledsession.RecoveryRetryCleanupV1 { + t.Fatalf("cleanup=%q, recovery=%q", cleanup.Kind, recovery) + } + if !supervisor.watchdogExitObserved || !errors.Is(supervisor.diagnosticErr, exitErr) || watchdog.disarmed || !watchdog.closed { + t.Fatalf("watchdog observed=%t, diagnostic=%v, disarmed=%t, closed=%t", + supervisor.watchdogExitObserved, supervisor.diagnosticErr, watchdog.disarmed, watchdog.closed) + } +} + func TestControlledSessionCleanupFailureRetainsDurableOwnership(t *testing.T) { plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) operation, err := deploy.AcquireOperationLock(t.Context(), plan.Workload.DeploymentDirectory) @@ -518,7 +744,7 @@ func TestControlledSessionCleanupFailureRetainsDurableOwnership(t *testing.T) { t.Fatalf("admission = %q, %v", status, err) } if _, err := operation.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFromPlanV1( - plan, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1, + plan, controlledSessionTestDockerEndpointV1, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1, )); err != nil { t.Fatal(err) } @@ -549,7 +775,7 @@ func TestFinishControlledSessionOwnershipReacquiresAfterReleaseAttempt(t *testin t.Fatalf("admission = %q, %v", status, err) } if _, err := operation.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFromPlanV1( - plan, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1, + plan, controlledSessionTestDockerEndpointV1, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1, )); err != nil { t.Fatal(err) } @@ -1297,25 +1523,74 @@ func (transport *fakeControlledSessionTransportV1) snapshotEvents() []controlled } type fakeControlledSessionChannelV1 struct { - transport controlledsession.ControllerTransportV1 - closed bool + transport controlledsession.ControllerTransportV1 + closed bool + blockClaim bool + claimStarted chan struct{} + claimOnce sync.Once } type fakeControlledSessionWatchdogV1 struct { + mu sync.Mutex + done chan struct{} + exitErr error + closed bool disarmed bool err error onDisarm func() } +func (watchdog *fakeControlledSessionWatchdogV1) Close() error { + watchdog.mu.Lock() + defer watchdog.mu.Unlock() + watchdog.closed = true + return nil +} + +func (watchdog *fakeControlledSessionWatchdogV1) Done() <-chan struct{} { + watchdog.mu.Lock() + defer watchdog.mu.Unlock() + if watchdog.done == nil { + watchdog.done = make(chan struct{}) + } + return watchdog.done +} + +func (watchdog *fakeControlledSessionWatchdogV1) ExitError() error { + watchdog.mu.Lock() + defer watchdog.mu.Unlock() + return watchdog.exitErr +} + +func (watchdog *fakeControlledSessionWatchdogV1) exitUnexpectedly(err error) { + watchdog.mu.Lock() + defer watchdog.mu.Unlock() + if watchdog.done == nil { + watchdog.done = make(chan struct{}) + } + watchdog.exitErr = err + close(watchdog.done) +} + func (watchdog *fakeControlledSessionWatchdogV1) Disarm(context.Context) error { watchdog.disarmed = true + _ = watchdog.Close() if watchdog.onDisarm != nil { watchdog.onDisarm() } return watchdog.err } -func (channel *fakeControlledSessionChannelV1) Claim(context.Context) (controlledsession.ControllerTransportV1, error) { +func (channel *fakeControlledSessionChannelV1) Claim(ctx context.Context) (controlledsession.ControllerTransportV1, error) { + channel.claimOnce.Do(func() { + if channel.claimStarted != nil { + close(channel.claimStarted) + } + }) + if channel.blockClaim { + <-ctx.Done() + return nil, ctx.Err() + } return channel.transport, nil } @@ -1332,6 +1607,7 @@ type fakeControlledSessionProcessV1 struct { cleaned bool cleanupAttempts int cleanupErrs []error + onCleanup func() } func (process *fakeControlledSessionProcessV1) ContainerID() string { @@ -1370,6 +1646,9 @@ func (process *fakeControlledSessionProcessV1) ForceStop(context.Context) error func (process *fakeControlledSessionProcessV1) Cleanup(context.Context) error { process.cleanupAttempts++ + if process.onCleanup != nil { + process.onCleanup() + } if len(process.cleanupErrs) > 0 { err := process.cleanupErrs[0] process.cleanupErrs = process.cleanupErrs[1:] diff --git a/internal/dockerdeploy/controlled_session_watchdog.go b/internal/dockerdeploy/controlled_session_watchdog.go index e6a9d8f4..8efb99d2 100644 --- a/internal/dockerdeploy/controlled_session_watchdog.go +++ b/internal/dockerdeploy/controlled_session_watchdog.go @@ -23,11 +23,15 @@ const ( ) type controlledSessionWatchdogRuntimeV1 interface { + Done() <-chan struct{} + ExitError() error + Close() error Disarm(context.Context) error } type controlledSessionWatchdogCleanupBackendV1 struct { currentBootSession func() (string, error) + bindDockerEndpoint func(string) error inspectContainer func(context.Context, string) (map[string]string, bool, error) removeContainer func(context.Context, string) error removeChannel func(string) error @@ -57,7 +61,7 @@ func runControlledSessionWatchdogV1( backend controlledSessionWatchdogCleanupBackendV1, ) error { if manifestReader == nil || liveness == nil || ready == nil || - backend.currentBootSession == nil || backend.inspectContainer == nil || + backend.currentBootSession == nil || backend.bindDockerEndpoint == nil || backend.inspectContainer == nil || backend.removeContainer == nil || backend.removeChannel == nil { return fmt.Errorf("watchdog backend is incomplete") } @@ -72,6 +76,9 @@ func runControlledSessionWatchdogV1( if err != nil { return err } + if err := backend.bindDockerEndpoint(manifest.DockerEndpoint); err != nil { + return fmt.Errorf("bind watchdog Docker endpoint: %w", err) + } if len(manifest.Networks) != 0 || len(manifest.Volumes) != 0 { return fmt.Errorf("cleanup manifest names resources unsupported by this watchdog") } @@ -167,11 +174,19 @@ func cleanupControlledSessionWatchdogContainerV1( } func productionControlledSessionWatchdogCleanupBackendV1() controlledSessionWatchdogCleanupBackendV1 { + var dockerRun commandRunner return controlledSessionWatchdogCleanupBackendV1{ currentBootSession: deploy.CurrentBootSessionIDV1, - inspectContainer: inspectControlledSessionWatchdogContainerV1, + bindDockerEndpoint: func(endpoint string) error { + var err error + dockerRun, err = commandRunnerForPinnedDockerEndpointV1(endpoint, runCommandWithoutDockerPreflight) + return err + }, + inspectContainer: func(ctx context.Context, containerID string) (map[string]string, bool, error) { + return inspectControlledSessionWatchdogContainerV1(ctx, containerID, dockerRun) + }, removeContainer: func(ctx context.Context, containerID string) error { - return runDockerCommand(CommandSpec{Name: "docker", Args: []string{"container", "rm", "--force", containerID}}, RunOptions{Context: ctx}) + return dockerRun(CommandSpec{Name: "docker", Args: []string{"container", "rm", "--force", containerID}}, RunOptions{Context: ctx}) }, removeChannel: removeControlledSessionChannelDirectoryV1, } @@ -190,9 +205,9 @@ func removeControlledSessionChannelDirectoryV1(path string) error { return nil } -func inspectControlledSessionWatchdogContainerV1(ctx context.Context, containerID string) (map[string]string, bool, error) { +func inspectControlledSessionWatchdogContainerV1(ctx context.Context, containerID string, run commandRunner) (map[string]string, bool, error) { var output bytes.Buffer - err := runDockerCommand(CommandSpec{Name: "docker", Args: []string{ + err := run(CommandSpec{Name: "docker", Args: []string{ "container", "inspect", "--format", "{{json .Id}} {{json .Config.Labels}}", containerID, }}, RunOptions{Context: ctx, Stdout: &output, Stderr: &output}) if err != nil { diff --git a/internal/dockerdeploy/controlled_session_watchdog_integration_linux_test.go b/internal/dockerdeploy/controlled_session_watchdog_integration_linux_test.go index 5ce91dc0..cb6dc47e 100644 --- a/internal/dockerdeploy/controlled_session_watchdog_integration_linux_test.go +++ b/internal/dockerdeploy/controlled_session_watchdog_integration_linux_test.go @@ -37,7 +37,11 @@ func proveControlledSessionWatchdogParentLossV1( if err := os.MkdirAll(plan.Channel.HostDirectory, 0o700); err != nil { t.Fatal(err) } - ownership := controlledSessionOwnershipFromPlanV1(plan, controllerID, workloadID) + dockerEndpoint, err := verifiedLocalDockerEndpointV1(ctx, controlledSessionCommandSpecV1(plan.Controller.Create), defaultDockerPreflightTimeout) + if err != nil { + t.Fatal(err) + } + ownership := controlledSessionOwnershipFromPlanV1(plan, dockerEndpoint, controllerID, workloadID) bootSession, err := deploy.CurrentBootSessionIDV1() if err != nil { t.Fatal(err) @@ -66,7 +70,8 @@ func proveControlledSessionWatchdogParentLossV1( t.Fatal(err) } select { - case err := <-watchdog.done: + case <-watchdog.exited: + err := watchdog.ExitError() if err != nil { t.Fatalf("watchdog parent-loss cleanup: %v", err) } diff --git a/internal/dockerdeploy/controlled_session_watchdog_linux.go b/internal/dockerdeploy/controlled_session_watchdog_linux.go index 84b960bf..1e64bfab 100644 --- a/internal/dockerdeploy/controlled_session_watchdog_linux.go +++ b/internal/dockerdeploy/controlled_session_watchdog_linux.go @@ -24,11 +24,39 @@ const ( var controlledSessionWatchdogExecutableV1 = os.Executable type controlledSessionWatchdogProcessV1 struct { - pid int - liveness *os.File - done chan error - once sync.Once - disarmErr error + pid int + liveness *os.File + exited chan struct{} + exitMu sync.Mutex + exitErr error + closeOnce sync.Once + closeErr error + disarmOnce sync.Once + disarmErr error +} + +func (watchdog *controlledSessionWatchdogProcessV1) Done() <-chan struct{} { + return watchdog.exited +} + +func (watchdog *controlledSessionWatchdogProcessV1) ExitError() error { + select { + case <-watchdog.exited: + watchdog.exitMu.Lock() + defer watchdog.exitMu.Unlock() + return watchdog.exitErr + default: + return nil + } +} + +func (watchdog *controlledSessionWatchdogProcessV1) Close() error { + watchdog.closeOnce.Do(func() { + if err := watchdog.liveness.Close(); err != nil { + watchdog.closeErr = fmt.Errorf("close controlled-session watchdog liveness pipe: %w", err) + } + }) + return watchdog.closeErr } func startControlledSessionWatchdogV1(ctx context.Context, manifest deploy.ControlledSessionCleanupManifest) (controlledSessionWatchdogRuntimeV1, error) { @@ -91,8 +119,16 @@ func startControlledSessionWatchdogV1(ctx context.Context, manifest deploy.Contr return nil, fmt.Errorf("launch controlled-session watchdog child: %w", err) } _ = readyWrite.Close() - done := make(chan error, 1) - go func() { done <- command.Wait() }() + watchdog := &controlledSessionWatchdogProcessV1{ + pid: command.Process.Pid, liveness: livenessWrite, exited: make(chan struct{}), + } + go func() { + err := command.Wait() + watchdog.exitMu.Lock() + watchdog.exitErr = err + watchdog.exitMu.Unlock() + close(watchdog.exited) + }() readyResult := make(chan error, 1) go func() { @@ -113,11 +149,12 @@ func startControlledSessionWatchdogV1(ctx context.Context, manifest deploy.Contr if err != nil { _ = livenessWrite.Close() _ = command.Process.Kill() - <-done + <-watchdog.exited return nil, fmt.Errorf("wait for controlled-session watchdog readiness: %w", err) } - case err := <-done: + case <-watchdog.exited: _ = livenessWrite.Close() + err := watchdog.ExitError() if err == nil { return nil, fmt.Errorf("controlled-session watchdog exited before readiness") } @@ -125,25 +162,24 @@ func startControlledSessionWatchdogV1(ctx context.Context, manifest deploy.Contr case <-ctx.Done(): _ = livenessWrite.Close() _ = command.Process.Kill() - <-done + <-watchdog.exited return nil, fmt.Errorf("wait for controlled-session watchdog readiness: %w", ctx.Err()) } - return &controlledSessionWatchdogProcessV1{pid: command.Process.Pid, liveness: livenessWrite, done: done}, nil + return watchdog, nil } func (watchdog *controlledSessionWatchdogProcessV1) Disarm(ctx context.Context) error { - watchdog.once.Do(func() { + watchdog.disarmOnce.Do(func() { if count, err := watchdog.liveness.Write([]byte{controlledSessionWatchdogDisarmByte}); err != nil || count != 1 { if err == nil { err = io.ErrShortWrite } watchdog.disarmErr = fmt.Errorf("send controlled-session watchdog disarm signal: %w", err) } - if err := watchdog.liveness.Close(); err != nil { - watchdog.disarmErr = errors.Join(watchdog.disarmErr, fmt.Errorf("close controlled-session watchdog liveness pipe: %w", err)) - } + watchdog.disarmErr = errors.Join(watchdog.disarmErr, watchdog.Close()) select { - case err := <-watchdog.done: + case <-watchdog.exited: + err := watchdog.ExitError() if err != nil { watchdog.disarmErr = errors.Join(watchdog.disarmErr, fmt.Errorf("wait for controlled-session watchdog child: %w", err)) } diff --git a/internal/dockerdeploy/controlled_session_watchdog_test.go b/internal/dockerdeploy/controlled_session_watchdog_test.go index c2ff1702..dbbf5b16 100644 --- a/internal/dockerdeploy/controlled_session_watchdog_test.go +++ b/internal/dockerdeploy/controlled_session_watchdog_test.go @@ -24,6 +24,7 @@ func TestControlledSessionWatchdogDisarmIndependentlyVerifiesCleanup(t *testing. channelVerified := false err = runControlledSessionWatchdogV1(bytes.NewReader(content), bytes.NewReader([]byte{controlledSessionWatchdogDisarmByte}), &ready, controlledSessionWatchdogCleanupBackendV1{ currentBootSession: func() (string, error) { return manifest.BootSession, nil }, + bindDockerEndpoint: func(string) error { return nil }, inspectContainer: func(context.Context, string) (map[string]string, bool, error) { inspectCount++ return nil, false, nil @@ -52,6 +53,10 @@ func TestControlledSessionWatchdogParentLossRemovesOnlyManifestResources(t *test var operations []string backend := controlledSessionWatchdogCleanupBackendV1{ currentBootSession: func() (string, error) { return manifest.BootSession, nil }, + bindDockerEndpoint: func(endpoint string) error { + operations = append(operations, "endpoint:"+endpoint) + return nil + }, inspectContainer: func(_ context.Context, id string) (map[string]string, bool, error) { operations = append(operations, "inspect:"+id) labels, found := containers[id] @@ -71,6 +76,7 @@ func TestControlledSessionWatchdogParentLossRemovesOnlyManifestResources(t *test t.Fatal(err) } want := []string{ + "endpoint:" + manifest.DockerEndpoint, "inspect:" + manifest.Workload.ID, "remove:" + manifest.Workload.ID, "inspect:" + manifest.Workload.ID, "inspect:" + manifest.Controller.ID, "remove:" + manifest.Controller.ID, "inspect:" + manifest.Controller.ID, "channel:" + manifest.ChannelDirectory, @@ -112,6 +118,7 @@ func TestControlledSessionWatchdogRejectsPriorBootBeforeReady(t *testing.T) { var ready bytes.Buffer err = runControlledSessionWatchdogV1(bytes.NewReader(content), strings.NewReader(""), &ready, controlledSessionWatchdogCleanupBackendV1{ currentBootSession: func() (string, error) { return "different-boot", nil }, + bindDockerEndpoint: func(string) error { return nil }, inspectContainer: func(context.Context, string) (map[string]string, bool, error) { return nil, false, errors.New("must not inspect") }, @@ -139,6 +146,7 @@ func TestControlledSessionWatchdogRechecksBootBeforeParentLossCleanup(t *testing } return "different-boot", nil }, + bindDockerEndpoint: func(string) error { return nil }, inspectContainer: func(context.Context, string) (map[string]string, bool, error) { cleanupCalled = true return nil, false, nil @@ -154,7 +162,7 @@ func TestControlledSessionWatchdogRechecksBootBeforeParentLossCleanup(t *testing func controlledSessionWatchdogManifestFixtureV1(t *testing.T) deploy.ControlledSessionCleanupManifest { t.Helper() plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) - ownership := controlledSessionOwnershipFromPlanV1(plan, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1) + ownership := controlledSessionOwnershipFromPlanV1(plan, controlledSessionTestDockerEndpointV1, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1) ownership.BootSession = "boot-session" manifest, err := deploy.ControlledSessionCleanupManifestFromOwnership(ownership) if err != nil { diff --git a/internal/dockerdeploy/controlled_session_workload_pty.go b/internal/dockerdeploy/controlled_session_workload_pty.go index 6dec9dec..9710cf8a 100644 --- a/internal/dockerdeploy/controlled_session_workload_pty.go +++ b/internal/dockerdeploy/controlled_session_workload_pty.go @@ -22,6 +22,7 @@ type dockerPTYAttachmentV1 interface { } type dockerWorkloadPTYBackendV1 struct { + bind func(context.Context, CommandSpec, time.Duration) (CommandSpec, commandRunner, error) run commandRunner recordContainerID func(string) error recordRollbackVerified func() @@ -40,6 +41,7 @@ type dockerWorkloadPTYWaitResultV1 struct { // attachment; container cleanup remains the lifecycle supervisor's job. type DockerWorkloadPTYV1 struct { plan ControlledSessionContainerPlanV1 + docker CommandSpec containerID string backend dockerWorkloadPTYBackendV1 attachment dockerPTYAttachmentV1 @@ -76,7 +78,7 @@ func prepareDockerWorkloadPTYWithContainerIDV1( recordRollbackVerified func(), ) (*DockerWorkloadPTYV1, error) { return prepareDockerWorkloadPTYV1(ctx, plan, dockerWorkloadPTYBackendV1{ - run: runDockerCommand, + bind: bindPinnedDockerCommandRunnerV1, recordContainerID: recordContainerID, recordRollbackVerified: recordRollbackVerified, attach: attachDockerContainerPTYV1, @@ -97,13 +99,23 @@ func prepareDockerWorkloadPTYV1( if plan.Role != ControlledSessionRoleWorkloadV1 { return nil, fmt.Errorf("prepare controlled-session workload PTY: container role must be %q", ControlledSessionRoleWorkloadV1) } - if backend.run == nil || backend.attach == nil || backend.resize == nil || backend.observe == nil { + if (backend.bind == nil && backend.run == nil) || backend.attach == nil || backend.resize == nil || backend.observe == nil { return nil, fmt.Errorf("prepare controlled-session workload PTY: backend is incomplete") } if ctx == nil { ctx = context.Background() } create := controlledSessionCommandSpecV1(plan.Create) + if backend.bind != nil { + var err error + create, backend.run, err = backend.bind(ctx, create, defaultDockerPreflightTimeout) + if err != nil { + return nil, fmt.Errorf("bind controlled-session workload Docker endpoint: %w", err) + } + if backend.run == nil { + return nil, fmt.Errorf("prepare controlled-session workload PTY: Docker endpoint binder returned no command runner") + } + } var createOutput bytes.Buffer var createErrorOutput bytes.Buffer if err := backend.run(create, RunOptions{Context: ctx, Stdout: &createOutput, Stderr: &createErrorOutput}); err != nil { @@ -144,7 +156,7 @@ func prepareDockerWorkloadPTYV1( return nil, attachErr } return &DockerWorkloadPTYV1{ - plan: plan, containerID: containerID, backend: backend, attachment: attachment, waitDone: make(chan struct{}), + plan: plan, docker: create, containerID: containerID, backend: backend, attachment: attachment, waitDone: make(chan struct{}), }, nil } @@ -192,7 +204,7 @@ func (workload *DockerWorkloadPTYV1) Start(ctx context.Context) error { if ctx == nil { ctx = context.Background() } - start := CommandSpec{Name: workload.plan.Start.Name, Args: []string{"start", workload.containerID}} + start := CommandSpec{Name: workload.docker.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() @@ -259,7 +271,7 @@ func (workload *DockerWorkloadPTYV1) resizeLocked(ctx context.Context, columns u } if err := workload.backend.resize( ctx, - controlledSessionCommandSpecV1(workload.plan.Start), + workload.docker, workload.containerID, columns, rows, @@ -291,7 +303,7 @@ func (workload *DockerWorkloadPTYV1) signal(ctx context.Context, signal string, if ctx == nil { ctx = context.Background() } - command := CommandSpec{Name: workload.plan.Start.Name, Args: []string{ + command := CommandSpec{Name: workload.docker.Name, Args: []string{ "kill", "--signal", signal, workload.containerID, }} if err := workload.backend.run(command, RunOptions{Context: ctx}); err != nil { @@ -336,7 +348,7 @@ func (workload *DockerWorkloadPTYV1) Cleanup(ctx context.Context) error { if ctx == nil { ctx = context.Background() } - cleanup := CommandSpec{Name: workload.plan.Cleanup.Name, Args: []string{"container", "rm", "--force", workload.containerID}} + cleanup := CommandSpec{Name: workload.docker.Name, Args: []string{"container", "rm", "--force", workload.containerID}} if err := workload.backend.run(cleanup, RunOptions{Context: ctx}); err != nil && !isMissingContainerCleanupError(err) { return fmt.Errorf("remove controlled-session workload container %q: %w", workload.plan.Container, err) } @@ -347,7 +359,7 @@ func (workload *DockerWorkloadPTYV1) Cleanup(ctx context.Context) error { } func (workload *DockerWorkloadPTYV1) observeExit() { - code, err := workload.backend.observe(context.Background(), controlledSessionCommandSpecV1(workload.plan.Start), workload.containerID) + code, err := workload.backend.observe(context.Background(), workload.docker, workload.containerID) if err != nil { workload.waitResult = dockerWorkloadPTYWaitResultV1{ status: controlledsession.ProcessStatusV1{ diff --git a/internal/dockerdeploy/controlled_session_workload_pty_test.go b/internal/dockerdeploy/controlled_session_workload_pty_test.go index dffffbef..627ba727 100644 --- a/internal/dockerdeploy/controlled_session_workload_pty_test.go +++ b/internal/dockerdeploy/controlled_session_workload_pty_test.go @@ -168,6 +168,78 @@ func TestDockerWorkloadPTYV1OrdersAttachStartResizeAndExactOperations(t *testing } } +func TestDockerWorkloadPTYV1PinsOneDockerEndpointForItsLifetime(t *testing.T) { + plan := controlledSessionWorkloadPlanFixtureV1(t) + const endpoint = "unix:///session-engine.sock" + var mu sync.Mutex + var commands []CommandSpec + record := func(spec CommandSpec) { + mu.Lock() + defer mu.Unlock() + commands = append(commands, spec) + } + binds := 0 + backend := dockerWorkloadPTYBackendV1{ + bind: func(_ context.Context, spec CommandSpec, _ time.Duration) (CommandSpec, commandRunner, error) { + binds++ + return pinDockerEndpointV1(spec, endpoint), func(command CommandSpec, options RunOptions) error { + command = pinDockerEndpointV1(command, endpoint) + record(command) + writeDockerWorkloadCreateIDV1(command, options, plan) + return nil + }, nil + }, + attach: func(_ context.Context, docker CommandSpec, _ string, _ time.Duration) (dockerPTYAttachmentV1, error) { + record(docker) + return &fakeDockerPTYAttachmentV1{}, nil + }, + resize: func(_ context.Context, docker CommandSpec, _ string, _, _ uint32, _ time.Duration) error { + record(docker) + return nil + }, + observe: func(_ context.Context, docker CommandSpec, _ string) (int, error) { + record(docker) + return 0, 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 { + t.Fatal(err) + } + if _, err := workload.Wait(t.Context()); err != nil { + t.Fatal(err) + } + if err := workload.RequestGracefulStop(t.Context()); err != nil { + t.Fatal(err) + } + if err := workload.ForceStop(t.Context()); err != nil { + t.Fatal(err) + } + if err := workload.Cleanup(t.Context()); err != nil { + t.Fatal(err) + } + mu.Lock() + got := append([]CommandSpec(nil), commands...) + mu.Unlock() + if binds != 1 || len(got) < 8 { + t.Fatalf("endpoint binds=%d commands=%#v", binds, got) + } + for _, command := range got { + if host, found := commandSpecEnvironmentValueV1(command, "DOCKER_HOST"); !found || host != endpoint { + t.Fatalf("command %#v used Docker host %q, found=%t", command.Args, host, found) + } + if contextName, found := commandSpecEnvironmentValueV1(command, "DOCKER_CONTEXT"); !found || contextName != "" { + t.Fatalf("command %#v retained Docker context %q, found=%t", command.Args, contextName, found) + } + } +} + func TestDockerWorkloadPTYV1TreatsMissingContainerAsCleaned(t *testing.T) { plan := controlledSessionWorkloadPlanFixtureV1(t) cleanupAttempts := 0 diff --git a/internal/dockerdeploy/live_run_recovery.go b/internal/dockerdeploy/live_run_recovery.go index 2e74a75b..2302c60b 100644 --- a/internal/dockerdeploy/live_run_recovery.go +++ b/internal/dockerdeploy/live_run_recovery.go @@ -16,6 +16,8 @@ import ( const liveRunRecoveryCleanupTimeoutV1 = 2 * time.Second +type legacyControlledSessionDockerEndpointResolverV1 func(context.Context) (string, error) + func recoverLiveRunQueueV1( ctx context.Context, operation *deploy.OperationLock, @@ -24,6 +26,9 @@ func recoverLiveRunQueueV1( ) (deploy.LiveRunRecoveryV1, error) { return recoverLiveRunQueueWithinV1( ctx, operation, notice, removeContainer, liveRunRecoveryCleanupTimeoutV1, + func(ctx context.Context) (string, error) { + return verifiedLocalDockerEndpointV1(ctx, CommandSpec{Name: "docker"}, defaultDockerPreflightTimeout) + }, ) } @@ -33,6 +38,7 @@ func recoverLiveRunQueueWithinV1( notice io.Writer, removeContainer commandRunner, cleanupTimeout time.Duration, + resolveLegacyDockerEndpoint legacyControlledSessionDockerEndpointResolverV1, ) (deploy.LiveRunRecoveryV1, error) { recovery, err := operation.RecoverLiveRunQueueV1() if err != nil { @@ -87,7 +93,9 @@ func recoverLiveRunQueueWithinV1( } break } - if err := cleanupControlledSessionRecoveryV1(cleanupContext, operation, ownership, removeContainer); err != nil { + if err := cleanupControlledSessionRecoveryV1( + cleanupContext, operation, ownership, removeContainer, resolveLegacyDockerEndpoint, + ); err != nil { if notice != nil { fmt.Fprintf(notice, "warning: deferred cleanup of recovered controlled session %q: %v\n", @@ -115,15 +123,31 @@ func cleanupControlledSessionRecoveryV1( operation *deploy.OperationLock, ownership deploy.ControlledSessionOwnershipV1, run commandRunner, + resolveLegacyDockerEndpoint legacyControlledSessionDockerEndpointResolverV1, ) error { deploymentDir := filepath.Dir(filepath.Dir(operation.Path())) expectedChannel := filepath.Join(deploymentDir, privateRuntimeMetadataDirectoryName, "sessions", ownership.LiveRunID) if ownership.ChannelDirectory != expectedChannel { return fmt.Errorf("refuse controlled-session recovery because channel directory %q is outside the exact deployment session path", ownership.ChannelDirectory) } + endpoint := ownership.DockerEndpoint + if endpoint == "" { + if resolveLegacyDockerEndpoint == nil { + return fmt.Errorf("resolve legacy controlled-session Docker endpoint: resolver is unavailable") + } + var err error + endpoint, err = resolveLegacyDockerEndpoint(ctx) + if err != nil { + return fmt.Errorf("resolve legacy controlled-session Docker endpoint: %w", err) + } + } + pinnedRun, err := commandRunnerForPinnedDockerEndpointV1(endpoint, run) + if err != nil { + return fmt.Errorf("bind recovered controlled-session Docker endpoint: %w", err) + } var cleanupErr error for _, container := range []deploy.ControlledSessionContainerOwnershipV1{ownership.Workload, ownership.Controller} { - cleanupErr = errors.Join(cleanupErr, cleanupControlledSessionRecoveryContainerV1(ctx, ownership.LiveRunID, container, run)) + cleanupErr = errors.Join(cleanupErr, cleanupControlledSessionRecoveryContainerV1(ctx, ownership.LiveRunID, container, pinnedRun)) } cleanupErr = errors.Join(cleanupErr, removeControlledSessionChannelDirectoryV1(ownership.ChannelDirectory)) return cleanupErr diff --git a/internal/dockerdeploy/live_run_recovery_test.go b/internal/dockerdeploy/live_run_recovery_test.go index db9ae4de..4e9a190c 100644 --- a/internal/dockerdeploy/live_run_recovery_test.go +++ b/internal/dockerdeploy/live_run_recovery_test.go @@ -105,6 +105,7 @@ func TestRecoverLiveRunQueueV1BoundsCleanupAcrossInventory(t *testing.T) { return options.Context.Err() }, 20*time.Millisecond, + nil, ) if err != nil { t.Fatal(err) @@ -138,7 +139,7 @@ func TestRecoverLiveRunQueueV1ReportsScheduledCleanupWhenDockerIsNotQueried(t *t } var notice bytes.Buffer if _, err := recoverLiveRunQueueWithinV1( - context.Background(), operation, ¬ice, nil, time.Second, + context.Background(), operation, ¬ice, nil, time.Second, nil, ); err != nil { t.Fatal(err) } @@ -189,7 +190,7 @@ func TestRecoverLiveRunQueueV1CleansPartialControlledSessionByVerifiedName(t *te if _, err := operation.AdmitLiveRunV1(run, false); err != nil { t.Fatal(err) } - ownership := controlledSessionOwnershipFromPlanV1(plan, "", "") + ownership := controlledSessionOwnershipFromPlanV1(plan, controlledSessionTestDockerEndpointV1, "", "") recorded, err := operation.RecordControlledSessionOwnershipV1(ownership) if err != nil { t.Fatal(err) @@ -219,6 +220,54 @@ func TestRecoverLiveRunQueueV1CleansPartialControlledSessionByVerifiedName(t *te } } +func TestRecoverLiveRunQueueV1CleansLegacyControlledSessionWithoutDockerEndpoint(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + operation, err := deploy.AcquireOperationLock(t.Context(), plan.Workload.DeploymentDirectory) + if err != nil { + t.Fatal(err) + } + defer operation.Unlock() + bootSession, err := deploy.CurrentBootSessionIDV1() + if err != nil { + t.Fatal(err) + } + ownership := controlledSessionOwnershipFromPlanV1( + plan, controlledSessionTestDockerEndpointV1, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1, + ) + ownership.BootSession = bootSession + ownership.DockerEndpoint = "" + queue := deploy.NewLiveRunQueueV1() + queue.ControlledSessions = []deploy.ControlledSessionOwnershipV1{ownership} + if err := operation.CommitLiveRunQueueV1(queue); err != nil { + t.Fatal(err) + } + if err := os.MkdirAll(ownership.ChannelDirectory, 0o700); err != nil { + t.Fatal(err) + } + containers := newControlledSessionRecoveryContainersV1(ownership) + containers.endpoint = controlledSessionTestDockerEndpointV1 + resolved := 0 + recovery, err := recoverLiveRunQueueWithinV1( + t.Context(), operation, nil, containers.run, liveRunRecoveryCleanupTimeoutV1, + func(context.Context) (string, error) { + resolved++ + return controlledSessionTestDockerEndpointV1, nil + }, + ) + if err != nil || len(recovery.ControlledSessions) != 1 { + t.Fatalf("legacy controlled-session recovery = %#v, error=%v", recovery, err) + } + if resolved != 1 { + t.Fatalf("legacy Docker endpoint resolutions = %d", resolved) + } + if len(containers.byID) != 0 { + t.Fatalf("legacy controlled-session containers remain = %#v", containers.byID) + } + if _, found, err := operation.ReadLiveRunQueueV1(); err != nil || found { + t.Fatalf("legacy queue remains: found=%t, error=%v", found, err) + } +} + func TestRecoverLiveRunQueueV1RetainsControlledSessionAfterLabelMismatchAndRetries(t *testing.T) { plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) operation, err := deploy.AcquireOperationLock(t.Context(), plan.Workload.DeploymentDirectory) @@ -232,7 +281,7 @@ func TestRecoverLiveRunQueueV1RetainsControlledSessionAfterLabelMismatchAndRetri if _, err := operation.AdmitLiveRunV1(run, false); err != nil { t.Fatal(err) } - ownership := controlledSessionOwnershipFromPlanV1(plan, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1) + ownership := controlledSessionOwnershipFromPlanV1(plan, controlledSessionTestDockerEndpointV1, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1) recorded, err := operation.RecordControlledSessionOwnershipV1(ownership) if err != nil { t.Fatal(err) @@ -270,12 +319,14 @@ type controlledSessionRecoveryContainersV1 struct { byID map[string]*controlledSessionRecoveryContainerFixtureV1 byName map[string]*controlledSessionRecoveryContainerFixtureV1 inspects []string + endpoint string } func newControlledSessionRecoveryContainersV1(ownership deploy.ControlledSessionOwnershipV1) *controlledSessionRecoveryContainersV1 { containers := &controlledSessionRecoveryContainersV1{ - byID: map[string]*controlledSessionRecoveryContainerFixtureV1{}, - byName: map[string]*controlledSessionRecoveryContainerFixtureV1{}, + byID: map[string]*controlledSessionRecoveryContainerFixtureV1{}, + byName: map[string]*controlledSessionRecoveryContainerFixtureV1{}, + endpoint: ownership.DockerEndpoint, } for _, input := range []struct { ownership deploy.ControlledSessionContainerOwnershipV1 @@ -301,6 +352,12 @@ func newControlledSessionRecoveryContainersV1(ownership deploy.ControlledSession } func (containers *controlledSessionRecoveryContainersV1) run(spec CommandSpec, options RunOptions) error { + if host, found := commandSpecEnvironmentValueV1(spec, "DOCKER_HOST"); !found || host != containers.endpoint { + return fmt.Errorf("controlled-session recovery command used Docker endpoint %q, want %q", host, containers.endpoint) + } + if contextName, found := commandSpecEnvironmentValueV1(spec, "DOCKER_CONTEXT"); !found || contextName != "" { + return fmt.Errorf("controlled-session recovery command retained Docker context %q", contextName) + } if len(spec.Args) >= 2 && spec.Args[0] == "container" && spec.Args[1] == "inspect" { target := spec.Args[len(spec.Args)-1] containers.inspects = append(containers.inspects, target) diff --git a/internal/dockerdeploy/live_runs_test.go b/internal/dockerdeploy/live_runs_test.go index 0ffc7c4b..3e441699 100644 --- a/internal/dockerdeploy/live_runs_test.go +++ b/internal/dockerdeploy/live_runs_test.go @@ -148,7 +148,7 @@ func TestStopLiveRunV1RemovesControlledSessionContainersAndRetainsOwnership(t *t t.Fatal(err) } ownership, err := operation.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFromPlanV1( - plan, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1, + plan, controlledSessionTestDockerEndpointV1, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1, )) if err != nil { t.Fatal(err) @@ -204,7 +204,7 @@ func TestStopLiveRunV1SkipsUnrecordedControlledSessionContainerIDs(t *testing.T) t.Fatal(err) } if _, err := operation.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFromPlanV1( - plan, dockerControllerTestContainerIDV1, "", + plan, controlledSessionTestDockerEndpointV1, dockerControllerTestContainerIDV1, "", )); err != nil { t.Fatal(err) } @@ -244,7 +244,7 @@ func TestStopLiveRunV1PreservesControlledSessionOnPartialCleanupFailure(t *testi t.Fatal(err) } ownership, err := operation.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFromPlanV1( - plan, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1, + plan, controlledSessionTestDockerEndpointV1, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1, )) if err != nil { t.Fatal(err)