From c851f67912bf0a01a36911f94d8d11032d1a2df1 Mon Sep 17 00:00:00 2001 From: Omry Yadan Date: Tue, 11 Aug 2026 14:12:13 +0800 Subject: [PATCH] Compose current controlled sessions Compose controlled sessions from exact current controller and workload generations under one live-run identity. Acquire both deployment locks and queue reservations atomically, enforce controller exclusivity, delegate the immutable plan to the supervisor, and recover abandoned ownership from either deployment. --- docs/CONTROLLED_SESSION_DESIGN.md | 7 +- internal/deploy/live_run_queue.go | 46 ++- internal/deploy/live_run_queue_file.go | 21 +- internal/deploy/live_run_queue_file_test.go | 10 +- .../controlled_session_supervisor.go | 102 ++++- .../current_controlled_session_run.go | 332 ++++++++++++++++ .../current_controlled_session_run_test.go | 359 ++++++++++++++++++ internal/dockerdeploy/live_run_recovery.go | 41 +- .../dockerdeploy/live_run_recovery_test.go | 148 +++++++- 9 files changed, 1034 insertions(+), 32 deletions(-) create mode 100644 internal/dockerdeploy/current_controlled_session_run.go create mode 100644 internal/dockerdeploy/current_controlled_session_run_test.go diff --git a/docs/CONTROLLED_SESSION_DESIGN.md b/docs/CONTROLLED_SESSION_DESIGN.md index e0f10ea3..afc416bb 100644 --- a/docs/CONTROLLED_SESSION_DESIGN.md +++ b/docs/CONTROLLED_SESSION_DESIGN.md @@ -128,7 +128,12 @@ summary: Capability-scoped execution sessions that inherit Reploy's global conta verified network absence after teardown. It also proves and documents the initial coarse boundary: either participant can reach any listening port on the other participant. Directional and per-port enforcement remain deferred - to the L3 gateway design. + to the L3 gateway design. Current-generation session composition now admits + the same live-run identity in both the controller and workload deployment + queues before startup. It records the exact session ownership in both queues + and removes both reservations only after verified cleanup, so an existing + controller run blocks admission and host-loss recovery from either + deployment fails closed until the abandoned session is verified absent. - Initial runtime: Linux containers under Docker - Motivating clients: OmegaFlow recording, sandboxed AI agents, security inspection, and untrusted-code execution diff --git a/internal/deploy/live_run_queue.go b/internal/deploy/live_run_queue.go index 0a75a79a..82094685 100644 --- a/internal/deploy/live_run_queue.go +++ b/internal/deploy/live_run_queue.go @@ -64,15 +64,17 @@ type LiveRunQueueV1 struct { } 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"` - NetworkID string `json:"network_id,omitempty"` - NetworkName string `json:"network_name,omitempty"` - Controller ControlledSessionContainerOwnershipV1 `json:"controller"` - Workload ControlledSessionContainerOwnershipV1 `json:"workload"` + LiveRunID string `json:"live_run_id"` + BootSession string `json:"boot_session"` + SessionHandle string `json:"session_handle"` + DockerEndpoint string `json:"docker_endpoint,omitempty"` + ControllerDeploymentDirectory string `json:"controller_deployment_directory,omitempty"` + WorkloadDeploymentDirectory string `json:"workload_deployment_directory,omitempty"` + ChannelDirectory string `json:"channel_directory"` + NetworkID string `json:"network_id,omitempty"` + NetworkName string `json:"network_name,omitempty"` + Controller ControlledSessionContainerOwnershipV1 `json:"controller"` + Workload ControlledSessionContainerOwnershipV1 `json:"workload"` } const ControlledSessionNetworkRoleV1 = "network" @@ -306,6 +308,29 @@ func validateControlledSessionOwnershipV1(ownership ControlledSessionOwnershipV1 return err } } + if (ownership.ControllerDeploymentDirectory == "") != (ownership.WorkloadDeploymentDirectory == "") { + return fmt.Errorf("controller and workload deployment directories must be recorded together") + } + if ownership.ControllerDeploymentDirectory != "" { + for _, participant := range []struct { + role string + directory string + }{ + {role: "controller", directory: ownership.ControllerDeploymentDirectory}, + {role: "workload", directory: ownership.WorkloadDeploymentDirectory}, + } { + if !filepath.IsAbs(participant.directory) || filepath.Clean(participant.directory) != participant.directory || !safeRecoveryIdentity(participant.directory) { + return fmt.Errorf("%s deployment directory must be a clean absolute path", participant.role) + } + } + if ownership.ControllerDeploymentDirectory == ownership.WorkloadDeploymentDirectory { + return fmt.Errorf("controller and workload deployment directories must be distinct") + } + expectedChannel := filepath.Join(ownership.WorkloadDeploymentDirectory, ".reploy", "sessions", ownership.LiveRunID) + if ownership.ChannelDirectory != expectedChannel { + return fmt.Errorf("channel directory must belong to the workload deployment") + } + } 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") } @@ -356,6 +381,9 @@ func validateCurrentControlledSessionOwnershipV1(ownership ControlledSessionOwne if ownership.DockerEndpoint == "" { return fmt.Errorf("Docker endpoint must be recorded for a new controlled session") } + if ownership.ControllerDeploymentDirectory == "" || ownership.WorkloadDeploymentDirectory == "" { + return fmt.Errorf("controller and workload deployment directories must be recorded for a new controlled session") + } return validateControlledSessionOwnershipV1(ownership) } diff --git a/internal/deploy/live_run_queue_file.go b/internal/deploy/live_run_queue_file.go index aaa6317f..a687642c 100644 --- a/internal/deploy/live_run_queue_file.go +++ b/internal/deploy/live_run_queue_file.go @@ -165,13 +165,28 @@ func (lock *OperationLock) RecordControlledSessionOwnershipV1(ownership Controll if admitted.Container != "" { return ControlledSessionOwnershipV1{}, fmt.Errorf("controlled session live run %q already names container %q", ownership.LiveRunID, admitted.Container) } - if admitted.GenerationReference != ownership.Workload.GenerationReference { - return ControlledSessionOwnershipV1{}, fmt.Errorf("controlled session workload generation does not match admitted live run %q", ownership.LiveRunID) - } ownership.BootSession = admitted.BootSession if err := validateCurrentControlledSessionOwnershipV1(ownership); err != nil { return ControlledSessionOwnershipV1{}, err } + deploymentDirectory := filepath.Dir(filepath.Dir(path)) + var participant ControlledSessionContainerOwnershipV1 + var role string + switch deploymentDirectory { + case ownership.ControllerDeploymentDirectory: + participant = ownership.Controller + role = "controller" + case ownership.WorkloadDeploymentDirectory: + participant = ownership.Workload + role = "workload" + default: + return ControlledSessionOwnershipV1{}, fmt.Errorf( + "controlled session ownership does not name deployment directory %q", deploymentDirectory, + ) + } + if admitted.GenerationReference != participant.GenerationReference { + return ControlledSessionOwnershipV1{}, fmt.Errorf("controlled session %s generation does not match admitted live run %q", role, ownership.LiveRunID) + } insert := sort.Search(len(queue.ControlledSessions), func(index int) bool { return queue.ControlledSessions[index].LiveRunID >= ownership.LiveRunID }) diff --git a/internal/deploy/live_run_queue_file_test.go b/internal/deploy/live_run_queue_file_test.go index 1cae9b7c..0faf0f79 100644 --- a/internal/deploy/live_run_queue_file_test.go +++ b/internal/deploy/live_run_queue_file_test.go @@ -300,10 +300,12 @@ 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"), + DockerEndpoint: "unix:///var/run/docker.sock", + ControllerDeploymentDirectory: dir + "-controller", + WorkloadDeploymentDirectory: dir, + 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/controlled_session_supervisor.go b/internal/dockerdeploy/controlled_session_supervisor.go index 156b9b04..087456a8 100644 --- a/internal/dockerdeploy/controlled_session_supervisor.go +++ b/internal/dockerdeploy/controlled_session_supervisor.go @@ -172,6 +172,21 @@ func RunControlledSessionV1( operation *deploy.OperationLock, plan ControlledSessionExecutionPlanV1, options ControlledSessionRunOptionsV1, +) (ControlledSessionRunResultV1, error) { + return runControlledSessionWithControllerReservationV1(ctx, operation, nil, plan, options) +} + +// runControlledSessionWithControllerReservationV1 additionally owns an +// admitted controller-side reservation when controllerOperation is non-nil. +// It records the same exact session ownership in both deployment queues before +// releasing either lock, so recovery from either participant must verify the +// old session absent before admitting replacement work. +func runControlledSessionWithControllerReservationV1( + ctx context.Context, + operation *deploy.OperationLock, + controllerOperation *deploy.OperationLock, + plan ControlledSessionExecutionPlanV1, + options ControlledSessionRunOptionsV1, ) (ControlledSessionRunResultV1, error) { if operation == nil { return ControlledSessionRunResultV1{}, fmt.Errorf("run controlled session requires an admitted operation lock") @@ -181,28 +196,46 @@ func RunControlledSessionV1( } absoluteDir, err := filepath.Abs(plan.Workload.DeploymentDirectory) if err != nil { - return ControlledSessionRunResultV1{}, releaseControlledSessionOperationV1(operation, fmt.Errorf("resolve controlled-session workload deployment directory: %w", err)) + return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionPairV1(operation, controllerOperation, plan.LiveRunID, fmt.Errorf("resolve controlled-session workload deployment directory: %w", err)) } if filepath.Dir(filepath.Dir(operation.Path())) != absoluteDir { - return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionV1(operation, plan.LiveRunID, fmt.Errorf("controlled-session operation lock does not belong to workload deployment %q", absoluteDir)) + return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionPairV1(operation, controllerOperation, plan.LiveRunID, fmt.Errorf("controlled-session operation lock does not belong to workload deployment %q", absoluteDir)) + } + if controllerOperation != nil { + if err := controllerOperation.RequireHeld(); err != nil { + return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionPairV1(operation, controllerOperation, plan.LiveRunID, err) + } + controllerDir, err := filepath.Abs(plan.Controller.DeploymentDirectory) + if err != nil { + return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionPairV1(operation, controllerOperation, plan.LiveRunID, fmt.Errorf("resolve controlled-session controller deployment directory: %w", err)) + } + if filepath.Dir(filepath.Dir(controllerOperation.Path())) != controllerDir { + return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionPairV1(operation, controllerOperation, plan.LiveRunID, fmt.Errorf("controlled-session controller operation lock does not belong to deployment %q", controllerDir)) + } } if ctx == nil || ctx.Done() == nil { - return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionV1(operation, plan.LiveRunID, fmt.Errorf("run controlled session: cancelable host context is required")) + return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionPairV1(operation, controllerOperation, plan.LiveRunID, fmt.Errorf("run controlled session: cancelable host context is required")) } if err := ValidateControlledSessionExecutionPlanV1(plan); err != nil { - return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionV1(operation, plan.LiveRunID, fmt.Errorf("run controlled session plan: %w", err)) + return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionPairV1(operation, controllerOperation, plan.LiveRunID, fmt.Errorf("run controlled session plan: %w", err)) } if err := validateControlledSessionRunOptionsV1(options); err != nil { - return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionV1(operation, plan.LiveRunID, err) + return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionPairV1(operation, controllerOperation, plan.LiveRunID, err) } if err := operation.RequireQueueEntryLeaseHeldV1(plan.LiveRunID); err != nil { - return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionV1(operation, plan.LiveRunID, fmt.Errorf("controlled-session admission ownership: %w", err)) + return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionPairV1(operation, controllerOperation, plan.LiveRunID, fmt.Errorf("controlled-session admission ownership: %w", err)) + } + if controllerOperation != nil { + if err := controllerOperation.RequireQueueEntryLeaseHeldV1(plan.LiveRunID); err != nil { + return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionPairV1(operation, controllerOperation, plan.LiveRunID, fmt.Errorf("controlled-session controller 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)) + return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionPairV1(operation, controllerOperation, plan.LiveRunID, fmt.Errorf("bind controlled-session Docker endpoint: %w", err)) } operationReleaseAttempted := false + controllerOperationReleaseAttempted := false ownershipRecorded := false partialPreparationCleanupVerified := false networkID := "" @@ -211,6 +244,11 @@ func RunControlledSessionV1( var incidentReceipt *deploy.ControlledSessionIncidentReceiptTargetV1 persistOwnership := func(controllerID string, workloadID string) (deploy.ControlledSessionOwnershipV1, error) { ownership := controlledSessionOwnershipWithNetworkFromPlanV1(plan, dockerEndpoint, networkID, controllerID, workloadID) + if controllerOperation != nil { + if _, err := controllerOperation.RecordControlledSessionOwnershipV1(ownership); err != nil { + return deploy.ControlledSessionOwnershipV1{}, fmt.Errorf("persist controlled-session controller ownership: %w", err) + } + } recorded, err := operation.RecordControlledSessionOwnershipV1(ownership) if err != nil { return deploy.ControlledSessionOwnershipV1{}, fmt.Errorf("persist controlled-session ownership: %w", err) @@ -299,6 +337,12 @@ func RunControlledSessionV1( if err := operation.Unlock(); err != nil { return deploy.ControlledSessionCleanupManifest{}, fmt.Errorf("release operation lock before controlled-session startup: %w", err) } + if controllerOperation != nil { + controllerOperationReleaseAttempted = true + if err := controllerOperation.Unlock(); err != nil { + return deploy.ControlledSessionCleanupManifest{}, fmt.Errorf("release controller operation lock before controlled-session startup: %w", err) + } + } return manifest, nil }, startWatchdog: func(ctx context.Context, manifest deploy.ControlledSessionCleanupManifest) (controlledSessionWatchdogRuntimeV1, error) { @@ -321,6 +365,13 @@ func RunControlledSessionV1( result.DeliveryTailCleanupStatus.Kind == controlledsession.CleanupStatusSucceededV1 cleaned = controlledSessionPreparationCanCompleteV1(cleaned, ownershipRecorded, operationReleaseAttempted, partialPreparationCleanupVerified) completionErr := finishControlledSessionOwnershipV1(context.WithoutCancel(ctx), absoluteDir, operation, operationReleaseAttempted, plan.LiveRunID, cleaned) + if controllerOperation != nil { + controllerCompletionErr := finishControlledSessionOwnershipV1( + context.WithoutCancel(ctx), plan.Controller.DeploymentDirectory, controllerOperation, + controllerOperationReleaseAttempted, plan.LiveRunID, cleaned, + ) + completionErr = errors.Join(completionErr, controllerCompletionErr) + } return result, errors.Join(runErr, completionErr) } @@ -364,9 +415,11 @@ func controlledSessionOwnershipWithNetworkFromPlanV1(plan ControlledSessionExecu } ownership := 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), + DockerEndpoint: dockerEndpoint, + ControllerDeploymentDirectory: plan.Controller.DeploymentDirectory, + WorkloadDeploymentDirectory: plan.Workload.DeploymentDirectory, + ChannelDirectory: plan.Channel.HostDirectory, + Controller: container(plan.Controller, controllerID), Workload: container(plan.Workload, workloadID), } if plan.Controller.SessionNetwork.Enabled { ownership.NetworkID = networkID @@ -412,6 +465,35 @@ func removeUnstartedControlledSessionV1(operation *deploy.OperationLock, runID s return releaseControlledSessionOperationV1(operation, errors.Join(cause, removeErr)) } +func removeUnstartedControlledSessionPairV1( + workload *deploy.OperationLock, + controller *deploy.OperationLock, + runID string, + cause error, +) error { + result := cause + for _, reservation := range []struct { + role string + operation *deploy.OperationLock + }{ + {role: "workload", operation: workload}, + {role: "controller", operation: controller}, + } { + if reservation.operation == nil { + continue + } + if deploy.ValidateLiveRunIDV1(runID) == nil { + if _, _, err := reservation.operation.RemoveLiveRunV1(runID); err != nil { + result = errors.Join(result, fmt.Errorf("remove unstarted controlled-session %s reservation: %w", reservation.role, err)) + } + } + if err := reservation.operation.Unlock(); err != nil { + result = errors.Join(result, fmt.Errorf("release controlled-session %s operation lock: %w", reservation.role, err)) + } + } + return result +} + func releaseControlledSessionOperationV1(operation *deploy.OperationLock, cause error) error { if err := operation.Unlock(); err != nil { return errors.Join(cause, fmt.Errorf("release controlled-session operation lock: %w", err)) diff --git a/internal/dockerdeploy/current_controlled_session_run.go b/internal/dockerdeploy/current_controlled_session_run.go new file mode 100644 index 00000000..4c03b0ab --- /dev/null +++ b/internal/dockerdeploy/current_controlled_session_run.go @@ -0,0 +1,332 @@ +package dockerdeploy + +import ( + "context" + "errors" + "fmt" + "io" + "os" + "sort" + + "github.com/omry/reploy/internal/blueprint" + "github.com/omry/reploy/internal/controlledsession" + "github.com/omry/reploy/internal/deploy" + "github.com/omry/reploy/internal/providerstore" +) + +// CurrentControlledSessionRunInputV1 selects two exact current generations and +// one declared controller command for a controlled session. Admission is +// intentionally immediate: a future queued form must preserve the same atomic +// two-deployment generation check while it waits. +type CurrentControlledSessionRunInputV1 struct { + ControllerDeploymentDir string + WorkloadDeploymentDir string + ControllerCommand string + ControllerArguments []string + EndpointIDs []string + InitialColumns uint32 + InitialRows uint32 + Runtime StagedProviderBuildRuntimeV1 + SupervisorOptions ControlledSessionRunOptionsV1 + Notice io.Writer +} + +type currentControlledSessionRuntimeV1 struct { + current CurrentBuild + plan CurrentRuntimePlanV1 +} + +type currentControlledSessionRunBackendV1 struct { + acquire func(context.Context, string) (*deploy.OperationLock, error) + loadRuntime func(context.Context, *deploy.OperationLock, string, StagedProviderBuildRuntimeV1) (currentControlledSessionRuntimeV1, error) + privateEnv func(string) (privateWorkloadEnvironmentV1, error) + concurrency func(blueprint.Document, DockerExecutionPlan, *transientOutputMount) (LiveRunConcurrencyDecisionV1, error) + newRunID func() (string, error) + newHandle func() (string, error) + acquireLease func(*deploy.OperationLock, string) (*deploy.QueueEntryLeaseV1, error) + await func(context.Context, string, *deploy.OperationLock, deploy.LiveRunV1, bool, io.Writer) (*deploy.OperationLock, error) + plan func(ControlledSessionPlanInputV1) (ControlledSessionExecutionPlanV1, error) + run func(context.Context, *deploy.OperationLock, *deploy.OperationLock, ControlledSessionExecutionPlanV1, ControlledSessionRunOptionsV1) (ControlledSessionRunResultV1, error) +} + +// RunCurrentControlledSessionV1 validates and admits one exact controller and +// workload pair before delegating all resource and lifecycle ownership to the +// controlled-session supervisor. It retains the controller deployment lock for +// the supervisor's complete lifetime, so another app, shell, control operation, +// or controlled session cannot use that controller generation concurrently. +func RunCurrentControlledSessionV1( + ctx context.Context, + input CurrentControlledSessionRunInputV1, +) (ControlledSessionRunResultV1, error) { + return runCurrentControlledSessionV1(ctx, input, currentControlledSessionRunBackendV1{ + acquire: deploy.AcquireOperationLock, + loadRuntime: loadCurrentControlledSessionRuntimeV1, + privateEnv: preparePrivateWorkloadEnvironmentV1, + concurrency: PlanLiveRunConcurrencyV1, + newRunID: deploy.NewLiveRunIDV1, + newHandle: controlledsession.NewHandleV1, + acquireLease: func(operation *deploy.OperationLock, id string) (*deploy.QueueEntryLeaseV1, error) { + return operation.AcquireLiveRunLeaseV1(id) + }, + await: AwaitLiveRunAdmissionWithNoticeV1, + plan: PlanControlledSessionV1, + run: runControlledSessionWithControllerReservationV1, + }) +} + +func runCurrentControlledSessionV1( + ctx context.Context, + input CurrentControlledSessionRunInputV1, + backend currentControlledSessionRunBackendV1, +) (result ControlledSessionRunResultV1, err error) { + if ctx == nil { + return result, fmt.Errorf("run current controlled session requires a context") + } + if err := ctx.Err(); err != nil { + return result, err + } + if backend.acquire == nil || backend.loadRuntime == nil || backend.privateEnv == nil || backend.concurrency == nil || backend.newRunID == nil || backend.newHandle == nil || backend.acquireLease == nil || backend.await == nil || backend.plan == nil || backend.run == nil { + return result, fmt.Errorf("run current controlled session requires a complete backend") + } + controllerDir, workloadDir, err := controlledSessionDeploymentDirectoriesV1(input.ControllerDeploymentDir, input.WorkloadDeploymentDir) + if err != nil { + return result, err + } + + lockDirectories := []string{controllerDir, workloadDir} + sort.Strings(lockDirectories) + operations := map[string]*deploy.OperationLock{} + runID := "" + for _, dir := range lockDirectories { + operation, acquireErr := backend.acquire(ctx, dir) + if acquireErr != nil { + return result, errors.Join(acquireErr, unlockControlledSessionOperationsV1(operations)) + } + operations[dir] = operation + } + defer func() { + if err != nil { + for _, operation := range operations { + if operation != nil && deploy.ValidateLiveRunIDV1(runID) == nil { + _, _, removeErr := operation.RemoveLiveRunV1(runID) + err = errors.Join(err, removeErr) + } + } + } + err = errors.Join(err, unlockControlledSessionOperationsV1(operations)) + }() + + controller, err := backend.loadRuntime(ctx, operations[controllerDir], controllerDir, input.Runtime) + if err != nil { + return result, fmt.Errorf("controlled-session controller runtime: %w", err) + } + workload, err := backend.loadRuntime(ctx, operations[workloadDir], workloadDir, input.Runtime) + if err != nil { + return result, fmt.Errorf("controlled-session workload runtime: %w", err) + } + for _, runtime := range []struct { + role string + dir string + }{ + {role: "controller", dir: controllerDir}, + {role: "workload", dir: workloadDir}, + } { + privateEnvironment, privateErr := backend.privateEnv(runtime.dir) + if privateErr != nil { + return result, fmt.Errorf("prepare controlled-session %s private environment: %w", runtime.role, privateErr) + } + if privateEnvironment.Present { + return result, fmt.Errorf("plan controlled session does not yet support private environment injection for the %s", runtime.role) + } + } + controllerConcurrency, err := backend.concurrency(controller.plan.Document, controller.plan.Docker, nil) + if err != nil { + return result, fmt.Errorf("plan controlled-session controller concurrency: %w", err) + } + workloadConcurrency, err := backend.concurrency(workload.plan.Document, workload.plan.Docker, nil) + if err != nil { + return result, fmt.Errorf("plan controlled-session workload concurrency: %w", err) + } + runID, err = backend.newRunID() + if err != nil { + return result, fmt.Errorf("create controlled-session live-run identity: %w", err) + } + leases := make([]*deploy.QueueEntryLeaseV1, 0, 2) + defer func() { + for _, lease := range leases { + if leaseErr := lease.Release(); leaseErr != nil { + err = errors.Join(err, fmt.Errorf("release controlled-session queue ownership: %w", leaseErr)) + } + } + }() + for _, reservation := range []struct { + role string + dir string + }{ + {role: "controller", dir: controllerDir}, + {role: "workload", dir: workloadDir}, + } { + lease, acquireErr := backend.acquireLease(operations[reservation.dir], runID) + if acquireErr != nil { + return result, fmt.Errorf("acquire controlled-session %s queue ownership: %w", reservation.role, acquireErr) + } + leases = append(leases, lease) + } + handle, err := backend.newHandle() + if err != nil { + return result, fmt.Errorf("create controlled-session handle: %w", err) + } + + plan, err := backend.plan(ControlledSessionPlanInputV1{ + Handle: handle, + LiveRunID: runID, + ControllerCurrent: controller.current, + ControllerRuntime: controller.plan, + ControllerCommand: input.ControllerCommand, + ControllerForwardedArguments: append([]string(nil), input.ControllerArguments...), + WorkloadCurrent: workload.current, + WorkloadRuntime: workload.plan, + EndpointIDs: append([]string(nil), input.EndpointIDs...), + InitialColumns: input.InitialColumns, + InitialRows: input.InitialRows, + }) + if err != nil { + return result, err + } + for _, admission := range []struct { + role string + dir string + current CurrentBuild + plan CurrentRuntimePlanV1 + concurrency LiveRunConcurrencyDecisionV1 + }{ + {role: "controller", dir: controllerDir, current: controller.current, plan: controller.plan, concurrency: controllerConcurrency}, + {role: "workload", dir: workloadDir, current: workload.current, plan: workload.plan, concurrency: workloadConcurrency}, + } { + candidate := deploy.LiveRunV1{ + ID: runID, + Kind: deploy.LiveRunKindShellV1, + Name: admission.plan.Document.Environment.ID, + GenerationReference: admission.current.Generation.Reference, + Exclusive: admission.role == "controller" || !admission.concurrency.AllowsOverlap, + WritableMount: admission.concurrency.WritableMount, + WritablePaths: admission.concurrency.WritablePaths, + } + operations[admission.dir], err = backend.await( + ctx, admission.dir, operations[admission.dir], candidate, false, input.Notice, + ) + if err != nil { + delete(operations, admission.dir) + return result, err + } + } + workloadOperation := operations[workloadDir] + controllerOperation := operations[controllerDir] + delete(operations, workloadDir) + delete(operations, controllerDir) + return backend.run(ctx, workloadOperation, controllerOperation, plan, input.SupervisorOptions) +} + +func controlledSessionDeploymentDirectoriesV1(controller string, workload string) (string, string, error) { + if controller == "" { + return "", "", fmt.Errorf("run current controlled session requires a controller deployment directory") + } + if workload == "" { + return "", "", fmt.Errorf("run current controlled session requires a workload deployment directory") + } + controllerDir, err := canonicalPathAllowMissingV1(controller) + if err != nil { + return "", "", fmt.Errorf("resolve controlled-session controller deployment directory: %w", err) + } + workloadDir, err := canonicalPathAllowMissingV1(workload) + if err != nil { + return "", "", fmt.Errorf("resolve controlled-session workload deployment directory: %w", err) + } + if controllerDir == workloadDir { + return "", "", fmt.Errorf("run current controlled session requires distinct controller and workload deployment directories") + } + if err := requireDistinctControlledSessionDeploymentFilesV1(controllerDir, workloadDir, os.Stat); err != nil { + return "", "", err + } + return controllerDir, workloadDir, nil +} + +func requireDistinctControlledSessionDeploymentFilesV1( + controllerDir string, + workloadDir string, + stat func(string) (os.FileInfo, error), +) error { + controllerInfo, err := stat(controllerDir) + if err != nil { + return fmt.Errorf("inspect controlled-session controller deployment directory: %w", err) + } + workloadInfo, err := stat(workloadDir) + if err != nil { + return fmt.Errorf("inspect controlled-session workload deployment directory: %w", err) + } + if os.SameFile(controllerInfo, workloadInfo) { + return fmt.Errorf("run current controlled session requires distinct controller and workload deployment directories") + } + return nil +} + +func loadCurrentControlledSessionRuntimeV1( + ctx context.Context, + operation *deploy.OperationLock, + dir string, + runtime StagedProviderBuildRuntimeV1, +) (currentControlledSessionRuntimeV1, error) { + store, err := providerstore.NewStore(dir) + if err != nil { + return currentControlledSessionRuntimeV1{}, err + } + state, found, err := operation.ReadStateV1() + if err != nil { + return currentControlledSessionRuntimeV1{}, err + } + if !found { + return currentControlledSessionRuntimeV1{}, fmt.Errorf("runtime state is missing; run `reploy stage` or `reploy install`") + } + document, err := blueprint.DecodeResolvedDocumentV1(state.Blueprint) + if err != nil { + return currentControlledSessionRuntimeV1{}, fmt.Errorf("runtime blueprint: %w", err) + } + current, found, err := ValidateCurrentBuild(ctx, operation, store, document.Environment.ID, dir) + if err != nil { + return currentControlledSessionRuntimeV1{}, fmt.Errorf("runtime current build: %w", err) + } + if !found { + return currentControlledSessionRuntimeV1{}, fmt.Errorf("%s", currentBuildRecoveryMessageV1(state, "runtime build is missing")) + } + planned, err := PlanCurrentRuntimeV1(CurrentRuntimePlanInputV1{ + DeploymentDir: dir, + Current: current, + Runtime: runtime, + }) + if err != nil { + return currentControlledSessionRuntimeV1{}, err + } + matched, err := CurrentBuildMatchesRuntimeV1(current, planned.Docker) + if err != nil { + return currentControlledSessionRuntimeV1{}, fmt.Errorf("runtime current-build check: %w", err) + } + if !matched { + return currentControlledSessionRuntimeV1{}, fmt.Errorf("%s", currentBuildRecoveryMessageV1(state, "runtime build is missing or stale")) + } + return currentControlledSessionRuntimeV1{current: current, plan: planned}, nil +} + +func unlockControlledSessionOperationsV1(operations map[string]*deploy.OperationLock) error { + directories := make([]string, 0, len(operations)) + for dir := range operations { + directories = append(directories, dir) + } + sort.Sort(sort.Reverse(sort.StringSlice(directories))) + var err error + for _, dir := range directories { + if operation := operations[dir]; operation != nil { + err = errors.Join(err, operation.Unlock()) + } + } + return err +} diff --git a/internal/dockerdeploy/current_controlled_session_run_test.go b/internal/dockerdeploy/current_controlled_session_run_test.go new file mode 100644 index 00000000..14c24e6d --- /dev/null +++ b/internal/dockerdeploy/current_controlled_session_run_test.go @@ -0,0 +1,359 @@ +package dockerdeploy + +import ( + "context" + "errors" + "io" + "os" + "path/filepath" + "reflect" + "strings" + "testing" + "time" + + "github.com/omry/reploy/internal/blueprint" + "github.com/omry/reploy/internal/deploy" +) + +func TestRunCurrentControlledSessionV1AdmitsExactPairAndDelegatesSupervisor(t *testing.T) { + root := t.TempDir() + controllerDir := filepath.Join(root, "z-controller") + workloadDir := filepath.Join(root, "a-workload") + for _, dir := range []string{controllerDir, workloadDir} { + if err := os.MkdirAll(dir, 0o700); err != nil { + t.Fatal(err) + } + } + fixture, planBackend := controlledSessionPlanFixtureV1(t) + fixture.ControllerRuntime.Docker.DeploymentDir = controllerDir + fixture.WorkloadRuntime.Docker.DeploymentDir = workloadDir + + var acquired []string + operations := map[string]*deploy.OperationLock{} + wantResult := ControlledSessionRunResultV1{ResultDelivered: true} + options := testControlledSessionRunOptionsV1() + backend := currentControlledSessionRunBackendV1{ + acquire: func(ctx context.Context, dir string) (*deploy.OperationLock, error) { + acquired = append(acquired, dir) + operation, err := deploy.AcquireOperationLock(ctx, dir) + if err == nil { + operations[dir] = operation + } + return operation, err + }, + loadRuntime: func(_ context.Context, operation *deploy.OperationLock, dir string, _ StagedProviderBuildRuntimeV1) (currentControlledSessionRuntimeV1, error) { + if err := operation.RequireHeld(); err != nil { + t.Fatalf("load runtime operation for %q is not held: %v", dir, err) + } + switch dir { + case controllerDir: + return currentControlledSessionRuntimeV1{current: fixture.ControllerCurrent, plan: fixture.ControllerRuntime}, nil + case workloadDir: + return currentControlledSessionRuntimeV1{current: fixture.WorkloadCurrent, plan: fixture.WorkloadRuntime}, nil + default: + t.Fatalf("unexpected runtime directory %q", dir) + return currentControlledSessionRuntimeV1{}, nil + } + }, + privateEnv: func(string) (privateWorkloadEnvironmentV1, error) { + return privateWorkloadEnvironmentV1{}, nil + }, + concurrency: func(_ blueprint.Document, _ DockerExecutionPlan, _ *transientOutputMount) (LiveRunConcurrencyDecisionV1, error) { + return LiveRunConcurrencyDecisionV1{AllowsOverlap: false, WritableMount: "workspace", WritablePaths: []string{"/workspace"}}, nil + }, + newRunID: func() (string, error) { return "run-0000000000000043", nil }, + newHandle: func() (string, error) { return "session-" + strings.Repeat("a", 64), nil }, + acquireLease: func(operation *deploy.OperationLock, id string) (*deploy.QueueEntryLeaseV1, error) { + return operation.AcquireLiveRunLeaseV1(id) + }, + await: func(_ context.Context, dir string, operation *deploy.OperationLock, candidate deploy.LiveRunV1, wait bool, _ io.Writer) (*deploy.OperationLock, error) { + if (dir != controllerDir && dir != workloadDir) || wait { + t.Fatalf("admission = dir %q wait %t", dir, wait) + } + if err := operations[controllerDir].RequireHeld(); err != nil { + t.Fatalf("controller generation lock was not held through admission: %v", err) + } + wantGeneration := fixture.WorkloadCurrent.Generation.Reference + if dir == controllerDir { + wantGeneration = fixture.ControllerCurrent.Generation.Reference + } + if candidate.GenerationReference != wantGeneration || !candidate.Exclusive || candidate.WritableMount != "workspace" || !reflect.DeepEqual(candidate.WritablePaths, []string{"/workspace"}) { + t.Fatalf("admission candidate = %#v", candidate) + } + status, err := operation.AdmitLiveRunV1(candidate, false) + if err != nil { + return nil, err + } + if status != deploy.LiveRunStatusActiveV1 { + t.Fatalf("admission status = %q", status) + } + return operation, nil + }, + plan: func(input ControlledSessionPlanInputV1) (ControlledSessionExecutionPlanV1, error) { + if input.Handle != "session-"+strings.Repeat("a", 64) || input.LiveRunID != "run-0000000000000043" || input.ControllerCommand != "inspect" || !reflect.DeepEqual(input.ControllerForwardedArguments, []string{"record"}) { + t.Fatalf("controlled-session plan input = %#v", input) + } + return planControlledSessionV1(input, planBackend) + }, + run: func(_ context.Context, operation *deploy.OperationLock, controllerOperation *deploy.OperationLock, plan ControlledSessionExecutionPlanV1, gotOptions ControlledSessionRunOptionsV1) (ControlledSessionRunResultV1, error) { + if err := operation.RequireHeld(); err != nil { + t.Fatalf("workload operation was not transferred held: %v", err) + } + if controllerOperation != operations[controllerDir] { + t.Fatal("controller operation was not transferred to supervisor") + } + if err := controllerOperation.RequireHeld(); err != nil { + t.Fatalf("controller lifetime reservation was not held: %v", err) + } + probeContext, cancel := context.WithTimeout(context.Background(), 25*time.Millisecond) + defer cancel() + probe, probeErr := deploy.AcquireOperationLock(probeContext, controllerDir) + if probe != nil { + _ = probe.Unlock() + } + if !errors.Is(probeErr, context.DeadlineExceeded) { + t.Fatalf("competing controller operation acquired during session: %v", probeErr) + } + if plan.Authorization.Handle != "session-"+strings.Repeat("a", 64) || plan.LiveRunID != "run-0000000000000043" || !reflect.DeepEqual(gotOptions, options) { + t.Fatalf("supervisor input = plan %#v options %#v", plan, gotOptions) + } + if _, removed, err := operation.RemoveLiveRunV1(plan.LiveRunID); err != nil || !removed { + t.Fatalf("remove admitted run = %t, %v", removed, err) + } + if _, removed, err := controllerOperation.RemoveLiveRunV1(plan.LiveRunID); err != nil || !removed { + t.Fatalf("remove controller reservation = %t, %v", removed, err) + } + if err := operation.Unlock(); err != nil { + t.Fatal(err) + } + if err := controllerOperation.Unlock(); err != nil { + t.Fatal(err) + } + return wantResult, nil + }, + } + + got, err := runCurrentControlledSessionV1(t.Context(), CurrentControlledSessionRunInputV1{ + ControllerDeploymentDir: controllerDir, + WorkloadDeploymentDir: workloadDir, + ControllerCommand: "inspect", + ControllerArguments: []string{"record"}, + InitialColumns: 100, + InitialRows: 28, + SupervisorOptions: options, + }, backend) + if err != nil { + t.Fatal(err) + } + if !reflect.DeepEqual(got, wantResult) { + t.Fatalf("result = %#v, want %#v", got, wantResult) + } + if !reflect.DeepEqual(acquired, []string{workloadDir, controllerDir}) { + t.Fatalf("operation lock order = %#v", acquired) + } + controllerOperation, err := deploy.AcquireOperationLock(t.Context(), controllerDir) + if err != nil { + t.Fatalf("controller reservation was not released after session: %v", err) + } + if err := controllerOperation.Unlock(); err != nil { + t.Fatal(err) + } +} + +func TestRunCurrentControlledSessionV1RejectsActiveControllerBeforeWorkloadAdmission(t *testing.T) { + root := t.TempDir() + controllerDir := filepath.Join(root, "controller") + workloadDir := filepath.Join(root, "workload") + for _, dir := range []string{controllerDir, workloadDir} { + if err := os.MkdirAll(dir, 0o700); err != nil { + t.Fatal(err) + } + } + fixture, _ := controlledSessionPlanFixtureV1(t) + fixture.ControllerRuntime.Docker.DeploymentDir = controllerDir + fixture.WorkloadRuntime.Docker.DeploymentDir = workloadDir + + controllerOperation, err := deploy.AcquireOperationLock(t.Context(), controllerDir) + if err != nil { + t.Fatal(err) + } + const existingID = "run-0000000000000042" + existingLease, err := controllerOperation.AcquireLiveRunLeaseV1(existingID) + if err != nil { + t.Fatal(err) + } + defer existingLease.Release() + if status, err := controllerOperation.AdmitLiveRunV1(deploy.LiveRunV1{ + ID: existingID, Kind: deploy.LiveRunKindAppV1, Name: "existing-controller", + GenerationReference: fixture.ControllerCurrent.Generation.Reference, + }, false); err != nil || status != deploy.LiveRunStatusActiveV1 { + t.Fatalf("existing controller admission = %q, %v", status, err) + } + if err := controllerOperation.Unlock(); err != nil { + t.Fatal(err) + } + + workloadAdmissionAttempted := false + _, err = runCurrentControlledSessionV1(t.Context(), CurrentControlledSessionRunInputV1{ + ControllerDeploymentDir: controllerDir, + WorkloadDeploymentDir: workloadDir, + }, currentControlledSessionRunBackendV1{ + acquire: deploy.AcquireOperationLock, + loadRuntime: func(_ context.Context, _ *deploy.OperationLock, dir string, _ StagedProviderBuildRuntimeV1) (currentControlledSessionRuntimeV1, error) { + if dir == controllerDir { + return currentControlledSessionRuntimeV1{current: fixture.ControllerCurrent, plan: fixture.ControllerRuntime}, nil + } + return currentControlledSessionRuntimeV1{current: fixture.WorkloadCurrent, plan: fixture.WorkloadRuntime}, nil + }, + privateEnv: func(string) (privateWorkloadEnvironmentV1, error) { return privateWorkloadEnvironmentV1{}, nil }, + concurrency: func(blueprint.Document, DockerExecutionPlan, *transientOutputMount) (LiveRunConcurrencyDecisionV1, error) { + return LiveRunConcurrencyDecisionV1{AllowsOverlap: true}, nil + }, + newRunID: func() (string, error) { return "run-0000000000000043", nil }, + newHandle: func() (string, error) { return "session-" + strings.Repeat("a", 64), nil }, + acquireLease: func(operation *deploy.OperationLock, id string) (*deploy.QueueEntryLeaseV1, error) { + return operation.AcquireLiveRunLeaseV1(id) + }, + await: func(ctx context.Context, dir string, operation *deploy.OperationLock, candidate deploy.LiveRunV1, wait bool, notice io.Writer) (*deploy.OperationLock, error) { + if dir == workloadDir { + workloadAdmissionAttempted = true + } + return AwaitLiveRunAdmissionWithNoticeV1(ctx, dir, operation, candidate, wait, notice) + }, + plan: func(ControlledSessionPlanInputV1) (ControlledSessionExecutionPlanV1, error) { + return ControlledSessionExecutionPlanV1{}, nil + }, + run: func(context.Context, *deploy.OperationLock, *deploy.OperationLock, ControlledSessionExecutionPlanV1, ControlledSessionRunOptionsV1) (ControlledSessionRunResultV1, error) { + return ControlledSessionRunResultV1{}, errors.New("must not run") + }, + }) + if !errors.Is(err, deploy.ErrLiveRunConflict) || workloadAdmissionAttempted { + t.Fatalf("controller conflict = workload attempted %t, error %v", workloadAdmissionAttempted, err) + } + + inspection, err := deploy.AcquireOperationLock(t.Context(), controllerDir) + if err != nil { + t.Fatal(err) + } + queue, _, err := inspection.ReadLiveRunQueueV1() + if err != nil || len(queue.Runs) != 1 || queue.Runs[0].ID != existingID { + t.Fatalf("controller queue = %#v, %v", queue, err) + } + if _, removed, err := inspection.RemoveLiveRunV1(existingID); err != nil || !removed { + t.Fatalf("remove existing controller = %t, %v", removed, err) + } + if err := inspection.Unlock(); err != nil { + t.Fatal(err) + } +} + +func TestRunCurrentControlledSessionV1RejectsOneDeploymentBeforeLocking(t *testing.T) { + dir := t.TempDir() + called := false + _, err := runCurrentControlledSessionV1(t.Context(), CurrentControlledSessionRunInputV1{ + ControllerDeploymentDir: dir, + WorkloadDeploymentDir: dir, + }, currentControlledSessionRunBackendV1{ + acquire: func(context.Context, string) (*deploy.OperationLock, error) { + called = true + return nil, nil + }, + loadRuntime: func(context.Context, *deploy.OperationLock, string, StagedProviderBuildRuntimeV1) (currentControlledSessionRuntimeV1, error) { + return currentControlledSessionRuntimeV1{}, nil + }, + privateEnv: func(string) (privateWorkloadEnvironmentV1, error) { + return privateWorkloadEnvironmentV1{}, nil + }, + concurrency: func(blueprint.Document, DockerExecutionPlan, *transientOutputMount) (LiveRunConcurrencyDecisionV1, error) { + return LiveRunConcurrencyDecisionV1{}, nil + }, + newRunID: func() (string, error) { return "", nil }, + newHandle: func() (string, error) { return "", nil }, + acquireLease: func(*deploy.OperationLock, string) (*deploy.QueueEntryLeaseV1, error) { return nil, nil }, + await: func(context.Context, string, *deploy.OperationLock, deploy.LiveRunV1, bool, io.Writer) (*deploy.OperationLock, error) { + return nil, nil + }, + plan: func(ControlledSessionPlanInputV1) (ControlledSessionExecutionPlanV1, error) { + return ControlledSessionExecutionPlanV1{}, nil + }, + run: func(context.Context, *deploy.OperationLock, *deploy.OperationLock, ControlledSessionExecutionPlanV1, ControlledSessionRunOptionsV1) (ControlledSessionRunResultV1, error) { + return ControlledSessionRunResultV1{}, nil + }, + }) + if err == nil || called { + t.Fatalf("same-directory result = called %t, error %v", called, err) + } +} + +func TestRunCurrentControlledSessionV1RejectsConfiguredPrivateEnvironmentBeforePlanning(t *testing.T) { + for _, configuredRole := range []string{"controller", "workload"} { + t.Run(configuredRole, func(t *testing.T) { + root := t.TempDir() + controllerDir := filepath.Join(root, "controller") + workloadDir := filepath.Join(root, "workload") + for _, dir := range []string{controllerDir, workloadDir} { + if err := os.MkdirAll(dir, 0o700); err != nil { + t.Fatal(err) + } + } + configuredDir := controllerDir + if configuredRole == "workload" { + configuredDir = workloadDir + } + if created, err := publishPrivateWorkloadEnvironmentFileV1( + filepath.Join(configuredDir, PrivateWorkloadEnvironmentFileName), + []byte("TOKEN=private\n"), + false, + ); err != nil || !created { + t.Fatalf("create private environment = %t, %v", created, err) + } + planned := false + _, err := runCurrentControlledSessionV1(t.Context(), CurrentControlledSessionRunInputV1{ + ControllerDeploymentDir: controllerDir, + WorkloadDeploymentDir: workloadDir, + }, currentControlledSessionRunBackendV1{ + acquire: deploy.AcquireOperationLock, + loadRuntime: func(context.Context, *deploy.OperationLock, string, StagedProviderBuildRuntimeV1) (currentControlledSessionRuntimeV1, error) { + return currentControlledSessionRuntimeV1{}, nil + }, + privateEnv: preparePrivateWorkloadEnvironmentV1, + concurrency: func(blueprint.Document, DockerExecutionPlan, *transientOutputMount) (LiveRunConcurrencyDecisionV1, error) { + return LiveRunConcurrencyDecisionV1{}, nil + }, + newRunID: func() (string, error) { return "run-0000000000000043", nil }, + newHandle: func() (string, error) { return "session-" + strings.Repeat("a", 64), nil }, + acquireLease: func(*deploy.OperationLock, string) (*deploy.QueueEntryLeaseV1, error) { + return nil, errors.New("must not acquire admission lease") + }, + await: func(context.Context, string, *deploy.OperationLock, deploy.LiveRunV1, bool, io.Writer) (*deploy.OperationLock, error) { + return nil, errors.New("must not admit") + }, + plan: func(ControlledSessionPlanInputV1) (ControlledSessionExecutionPlanV1, error) { + planned = true + return ControlledSessionExecutionPlanV1{}, nil + }, + run: func(context.Context, *deploy.OperationLock, *deploy.OperationLock, ControlledSessionExecutionPlanV1, ControlledSessionRunOptionsV1) (ControlledSessionRunResultV1, error) { + return ControlledSessionRunResultV1{}, errors.New("must not run") + }, + }) + if err == nil || !strings.Contains(err.Error(), "private environment injection for the "+configuredRole) || planned { + t.Fatalf("private-environment result = planned %t, error %v", planned, err) + } + }) + } +} + +func TestRequireDistinctControlledSessionDeploymentFilesV1RejectsFilesystemAlias(t *testing.T) { + directory := t.TempDir() + info, err := os.Stat(directory) + if err != nil { + t.Fatal(err) + } + err = requireDistinctControlledSessionDeploymentFilesV1( + "/controller", "/CONTROLLER", + func(string) (os.FileInfo, error) { return info, nil }, + ) + if err == nil || !strings.Contains(err.Error(), "requires distinct") { + t.Fatalf("filesystem-alias error = %v", err) + } +} diff --git a/internal/dockerdeploy/live_run_recovery.go b/internal/dockerdeploy/live_run_recovery.go index edf593da..6ccff65e 100644 --- a/internal/dockerdeploy/live_run_recovery.go +++ b/internal/dockerdeploy/live_run_recovery.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "io" + "os" "path/filepath" "reflect" "strings" @@ -87,8 +88,10 @@ func recoverLiveRunQueueWithinV1( ) } } + var controlledSessionCleanupErr error for _, ownership := range recovery.ControlledSessions { if err := cleanupContext.Err(); err != nil { + controlledSessionCleanupErr = errors.Join(controlledSessionCleanupErr, err) if notice != nil { fmt.Fprintf(notice, "warning: deferred remaining recovered controlled-session cleanup: %v\n", err) } @@ -97,6 +100,7 @@ func recoverLiveRunQueueWithinV1( if err := cleanupControlledSessionRecoveryV1( cleanupContext, operation, ownership, removeContainer, resolveLegacyDockerEndpoint, ); err != nil { + controlledSessionCleanupErr = errors.Join(controlledSessionCleanupErr, err) if notice != nil { fmt.Fprintf(notice, "warning: deferred cleanup of recovered controlled session %q: %v\n", @@ -116,6 +120,9 @@ func recoverLiveRunQueueWithinV1( ) } } + if controlledSessionCleanupErr != nil { + return recovery, fmt.Errorf("controlled-session recovery remains incomplete: %w", controlledSessionCleanupErr) + } return recovery, nil } @@ -127,7 +134,27 @@ func cleanupControlledSessionRecoveryV1( resolveLegacyDockerEndpoint legacyControlledSessionDockerEndpointResolverV1, ) error { deploymentDir := filepath.Dir(filepath.Dir(operation.Path())) - expectedChannel := filepath.Join(deploymentDir, privateRuntimeMetadataDirectoryName, "sessions", ownership.LiveRunID) + ownerDeploymentDir := deploymentDir + if ownership.WorkloadDeploymentDirectory != "" { + controllerParticipant := deploymentDir == ownership.ControllerDeploymentDirectory + workloadParticipant := deploymentDir == ownership.WorkloadDeploymentDirectory + if !controllerParticipant && !workloadParticipant { + var err error + controllerParticipant, err = sameControlledSessionDeploymentV1(deploymentDir, ownership.ControllerDeploymentDirectory) + if err != nil { + return fmt.Errorf("verify controlled-session controller participant: %w", err) + } + workloadParticipant, err = sameControlledSessionDeploymentV1(deploymentDir, ownership.WorkloadDeploymentDirectory) + if err != nil { + return fmt.Errorf("verify controlled-session workload participant: %w", err) + } + } + if controllerParticipant == workloadParticipant { + return fmt.Errorf("refuse controlled-session recovery because deployment %q is not an exact session participant", deploymentDir) + } + ownerDeploymentDir = ownership.WorkloadDeploymentDirectory + } + expectedChannel := filepath.Join(ownerDeploymentDir, 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) } @@ -157,6 +184,18 @@ func cleanupControlledSessionRecoveryV1( return cleanupErr } +func sameControlledSessionDeploymentV1(actual string, expected string) (bool, error) { + actualInfo, err := os.Stat(actual) + if err != nil { + return false, fmt.Errorf("inspect recovered deployment %q: %w", actual, err) + } + expectedInfo, err := os.Stat(expected) + if err != nil { + return false, fmt.Errorf("inspect recorded deployment %q: %w", expected, err) + } + return os.SameFile(actualInfo, expectedInfo), nil +} + func cleanupControlledSessionRecoveryNetworkV1( ctx context.Context, ownership deploy.ControlledSessionOwnershipV1, diff --git a/internal/dockerdeploy/live_run_recovery_test.go b/internal/dockerdeploy/live_run_recovery_test.go index 899341ce..aef16f63 100644 --- a/internal/dockerdeploy/live_run_recovery_test.go +++ b/internal/dockerdeploy/live_run_recovery_test.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "os" + "path/filepath" "reflect" "strings" "testing" @@ -71,6 +72,141 @@ func TestRecoverLiveRunQueueV1DefersAndRetriesExactContainerCleanup(t *testing.T } } +func TestRecoverLiveRunQueueV1BlocksControllerAdmissionUntilSessionCleanupVerifiesAbsent(t *testing.T) { + input, planBackend := controlledSessionPlanFixtureV1(t) + plan, err := planControlledSessionV1(input, planBackend) + if err != nil { + t.Fatal(err) + } + operations := map[string]*deploy.OperationLock{} + leases := map[string]*deploy.QueueEntryLeaseV1{} + for _, participant := range []struct { + dir string + name string + generation string + }{ + {dir: plan.Controller.DeploymentDirectory, name: plan.Controller.DeploymentID, generation: plan.Controller.GenerationReference}, + {dir: plan.Workload.DeploymentDirectory, name: plan.Workload.DeploymentID, generation: plan.Workload.GenerationReference}, + } { + operation, err := deploy.AcquireOperationLock(t.Context(), participant.dir) + if err != nil { + t.Fatal(err) + } + operations[participant.dir] = operation + lease, err := operation.AcquireLiveRunLeaseV1(plan.LiveRunID) + if err != nil { + t.Fatal(err) + } + leases[participant.dir] = lease + if status, err := operation.AdmitLiveRunV1(deploy.LiveRunV1{ + ID: plan.LiveRunID, Kind: deploy.LiveRunKindShellV1, Name: participant.name, + GenerationReference: participant.generation, Exclusive: true, + }, false); err != nil || status != deploy.LiveRunStatusActiveV1 { + t.Fatalf("participant admission = %q, %v", status, err) + } + } + ownership := controlledSessionOwnershipFromPlanV1(plan, "unix:///var/run/docker.sock", "", "") + for _, dir := range []string{plan.Controller.DeploymentDirectory, plan.Workload.DeploymentDirectory} { + if _, err := operations[dir].RecordControlledSessionOwnershipV1(ownership); err != nil { + t.Fatalf("record ownership in %q: %v", dir, err) + } + if err := operations[dir].Unlock(); err != nil { + t.Fatal(err) + } + if err := leases[dir].Release(); err != nil { + t.Fatal(err) + } + } + + controllerOperation, err := deploy.AcquireOperationLock(t.Context(), plan.Controller.DeploymentDirectory) + if err != nil { + t.Fatal(err) + } + want := errors.New("Docker cleanup failed") + if _, err := recoverLiveRunQueueWithinV1( + t.Context(), controllerOperation, nil, + func(CommandSpec, RunOptions) error { return want }, + time.Second, nil, + ); !errors.Is(err, want) || !strings.Contains(err.Error(), "recovery remains incomplete") { + t.Fatalf("failed controller recovery error = %v", err) + } + queue, _, err := controllerOperation.ReadLiveRunQueueV1() + if err != nil || len(queue.Runs) != 0 || len(queue.ControlledSessions) != 1 { + t.Fatalf("retained controller reservation = %#v, %v", queue, err) + } + + if _, err := recoverLiveRunQueueWithinV1( + t.Context(), controllerOperation, nil, + func(spec CommandSpec, options RunOptions) error { + if len(spec.Args) >= 2 && spec.Args[0] == "container" && spec.Args[1] == "inspect" { + _, _ = fmt.Fprintln(options.Stderr, "Error: No such container") + return errors.New("inspect failed") + } + return nil + }, + time.Second, nil, + ); err != nil { + t.Fatal(err) + } + queue, found, err := controllerOperation.ReadLiveRunQueueV1() + if err != nil || found || len(queue.ControlledSessions) != 0 { + t.Fatalf("completed controller recovery = %#v, found=%t, %v", queue, found, err) + } + if err := controllerOperation.Unlock(); err != nil { + t.Fatal(err) + } +} + +func TestCleanupControlledSessionRecoveryV1AcceptsParticipantFilesystemAlias(t *testing.T) { + input, planBackend := controlledSessionPlanFixtureV1(t) + plan, err := planControlledSessionV1(input, planBackend) + if err != nil { + t.Fatal(err) + } + operation, err := deploy.AcquireOperationLock(t.Context(), plan.Controller.DeploymentDirectory) + if err != nil { + t.Fatal(err) + } + defer operation.Unlock() + + alias := filepath.Join(t.TempDir(), "controller-alias") + if err := os.Symlink(plan.Controller.DeploymentDirectory, alias); err != nil { + t.Skipf("create deployment alias: %v", err) + } + ownership := controlledSessionOwnershipFromPlanV1(plan, controlledSessionTestDockerEndpointV1, "", "") + ownership.ControllerDeploymentDirectory = alias + resources := newControlledSessionRecoveryContainersV1(ownership) + if err := cleanupControlledSessionRecoveryV1(t.Context(), operation, ownership, resources.run, nil); err != nil { + t.Fatal(err) + } + if len(resources.byID) != 0 { + t.Fatalf("controlled-session containers remain = %#v", resources.byID) + } +} + +func TestCleanupControlledSessionRecoveryV1DoesNotRequireOtherParticipant(t *testing.T) { + input, planBackend := controlledSessionPlanFixtureV1(t) + plan, err := planControlledSessionV1(input, planBackend) + if err != nil { + t.Fatal(err) + } + operation, err := deploy.AcquireOperationLock(t.Context(), plan.Workload.DeploymentDirectory) + if err != nil { + t.Fatal(err) + } + defer operation.Unlock() + + ownership := controlledSessionOwnershipFromPlanV1(plan, controlledSessionTestDockerEndpointV1, "", "") + ownership.ControllerDeploymentDirectory = filepath.Join(t.TempDir(), "missing-controller") + resources := newControlledSessionRecoveryContainersV1(ownership) + if err := cleanupControlledSessionRecoveryV1(t.Context(), operation, ownership, resources.run, nil); err != nil { + t.Fatal(err) + } + if len(resources.byID) != 0 { + t.Fatalf("controlled-session containers remain = %#v", resources.byID) + } +} + func TestRecoverLiveRunQueueV1BoundsCleanupAcrossInventory(t *testing.T) { dir := t.TempDir() operation, err := deploy.AcquireOperationLock(t.Context(), dir) @@ -337,8 +473,10 @@ func TestRecoverLiveRunQueueV1RetainsControlledSessionAfterLabelMismatchAndRetri containers := newControlledSessionRecoveryContainersV1(recorded) containers.byID[dockerWorkloadTestContainerIDV1].labels["io.reploy.session.live-run"] = "run-ffffffffffffffff" var notice bytes.Buffer - if _, err := recoverLiveRunQueueV1(t.Context(), operation, ¬ice, containers.run); err != nil { - t.Fatal(err) + if _, err := recoverLiveRunQueueV1(t.Context(), operation, ¬ice, containers.run); err == nil || + !strings.Contains(err.Error(), "recovery remains incomplete") || + !strings.Contains(err.Error(), "ownership label") { + t.Fatalf("label-mismatch recovery error = %v", err) } queue, found, err := operation.ReadLiveRunQueueV1() if err != nil || !found || len(queue.Runs) != 0 || len(queue.ControlledSessions) != 1 || queue.ControlledSessions[0] != recorded { @@ -419,8 +557,10 @@ func TestRecoverLiveRunQueueV1RetainsLabelMismatchedNetworkAndRetries(t *testing resources := newControlledSessionRecoveryContainersV1(recorded) resources.network.labels["io.reploy.session.live-run"] = "run-ffffffffffffffff" var notice bytes.Buffer - if _, err := recoverLiveRunQueueV1(t.Context(), operation, ¬ice, resources.run); err != nil { - t.Fatal(err) + if _, err := recoverLiveRunQueueV1(t.Context(), operation, ¬ice, resources.run); err == nil || + !strings.Contains(err.Error(), "recovery remains incomplete") || + !strings.Contains(err.Error(), "ownership labels") { + t.Fatalf("network-mismatch recovery error = %v", err) } if resources.network == nil || !strings.Contains(notice.String(), "network") || !strings.Contains(notice.String(), "ownership labels") { t.Fatalf("mismatched network was not retained: network=%#v notice=%q", resources.network, notice.String())