Skip to content

Compose current controlled sessions #43

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
omry wants to merge 1 commit into from
Open

Compose current controlled sessions #43

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
7 changes: 6 additions & 1 deletion docs/CONTROLLED_SESSION_DESIGN.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
46 changes: 37 additions & 9 deletions internal/deploy/live_run_queue.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"
Expand Down Expand Up @@ -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")
}
Expand Down Expand Up @@ -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)
}

Expand Down
21 changes: 18 additions & 3 deletions internal/deploy/live_run_queue_file.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
})
Expand Down
10 changes: 6 additions & 4 deletions internal/deploy/live_run_queue_file_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"),
}
}

Expand Down
102 changes: 92 additions & 10 deletions internal/dockerdeploy/controlled_session_supervisor.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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")
Expand All @@ -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 := ""
Expand All @@ -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)
Expand Down Expand Up @@ -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) {
Expand All @@ -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)
}

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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))
Expand Down
Loading
Loading