From f62b8dba27124b2c0d5c1edf7eaa2a6391bb3efa Mon Sep 17 00:00:00 2001 From: Omry Yadan Date: Sun, 9 Aug 2026 17:32:33 +0800 Subject: [PATCH] Record controlled-session resource ownership --- docs/CONTROLLED_SESSION_DESIGN.md | 9 +- internal/deploy/live_run_queue.go | 94 +++++++++++++- internal/deploy/live_run_queue_file.go | 104 ++++++++++++++- internal/deploy/live_run_queue_file_test.go | 120 ++++++++++++++++++ .../controlled_session_controller.go | 4 + .../controlled_session_supervisor.go | 116 ++++++++++++++++- ...led_session_supervisor_integration_test.go | 26 +++- .../controlled_session_supervisor_test.go | 113 +++++++++++++++++ .../controlled_session_workload_pty.go | 4 + 9 files changed, 575 insertions(+), 15 deletions(-) diff --git a/docs/CONTROLLED_SESSION_DESIGN.md b/docs/CONTROLLED_SESSION_DESIGN.md index b1c944b9..1f7a835a 100644 --- a/docs/CONTROLLED_SESSION_DESIGN.md +++ b/docs/CONTROLLED_SESSION_DESIGN.md @@ -55,8 +55,13 @@ summary: Capability-scoped execution sessions that inherit Reploy's global conta completion and result acknowledgement, and removes both containers and the private channel. A workload that starts before a later startup step fails is still terminated and its output is finalized through the same barrier. - Crash watchdogs and restart reconciliation remain the next ownership phase; - controlled-session networking remains a later phase. + Before either process starts, the exact full controller and workload + container IDs, their planned ownership identities, the private channel + directory, and the session, lease, and boot identities are now durably + recorded in the existing live-run state. Verified cleanup removes that + record; failed cleanup retains it. The watchdog and restart reconciliation + remain the next 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 diff --git a/internal/deploy/live_run_queue.go b/internal/deploy/live_run_queue.go index 14e7eb2b..d82070b2 100644 --- a/internal/deploy/live_run_queue.go +++ b/internal/deploy/live_run_queue.go @@ -7,6 +7,7 @@ import ( "errors" "fmt" "io" + "path/filepath" "regexp" "github.com/omry/reploy/internal/canonical" @@ -55,9 +56,28 @@ type LiveRunV1 struct { } type LiveRunQueueV1 struct { - Schema string `json:"schema"` - Runs []LiveRunV1 `json:"runs"` - Cleanup []LiveRunContainerCleanupV1 `json:"cleanup,omitempty"` + Schema string `json:"schema"` + Runs []LiveRunV1 `json:"runs"` + ControlledSessions []ControlledSessionOwnershipV1 `json:"controlled_sessions,omitempty"` + Cleanup []LiveRunContainerCleanupV1 `json:"cleanup,omitempty"` +} + +type ControlledSessionOwnershipV1 struct { + LiveRunID string `json:"live_run_id"` + BootSession string `json:"boot_session"` + SessionHandle string `json:"session_handle"` + ChannelDirectory string `json:"channel_directory"` + Controller ControlledSessionContainerOwnershipV1 `json:"controller"` + Workload ControlledSessionContainerOwnershipV1 `json:"workload"` +} + +type ControlledSessionContainerOwnershipV1 struct { + Role string `json:"role"` + ID string `json:"id"` + Name string `json:"name"` + DeploymentID string `json:"deployment_id"` + GenerationReference string `json:"generation_reference"` + BuildIdentity string `json:"build_identity"` } type LiveRunRecoveryReasonV1 string @@ -98,6 +118,9 @@ var ErrLiveRunConflict = errors.New("another run must finish first") var liveRunIDPatternV1 = regexp.MustCompile(`^run-[0-9a-f]{16}$`) var controlMarkerIDPatternV1 = regexp.MustCompile(`^control-[0-9a-f]{16}$`) +var controlledSessionHandlePatternV1 = regexp.MustCompile(`^session-[0-9a-f]{64}$`) +var controlledSessionContainerIDPatternV1 = regexp.MustCompile(`^[0-9a-f]{64}$`) +var controlledSessionBuildIdentityPatternV1 = regexp.MustCompile(`^sha256:[0-9a-f]{64}$`) func NewLiveRunQueueV1() LiveRunQueueV1 { return LiveRunQueueV1{Schema: LiveRunQueueSchemaV1, Runs: []LiveRunV1{}} @@ -189,6 +212,14 @@ func ValidateLiveRunQueueV1(queue LiveRunQueueV1) error { if queue.Runs == nil { return fmt.Errorf("live run queue runs must use an array") } + for index, ownership := range queue.ControlledSessions { + if err := validateControlledSessionOwnershipV1(ownership); err != nil { + return fmt.Errorf("live run queue controlled session %d: %w", index, err) + } + if index > 0 && queue.ControlledSessions[index-1].LiveRunID >= ownership.LiveRunID { + return fmt.Errorf("live run queue controlled sessions must be sorted and unique by live run ID") + } + } for index, cleanup := range queue.Cleanup { if err := validateLiveRunContainerCleanupV1(cleanup); err != nil { return fmt.Errorf("live run queue cleanup entry %d: %w", index, err) @@ -247,6 +278,52 @@ func ValidateLiveRunQueueV1(queue LiveRunQueueV1) error { return nil } +func validateControlledSessionOwnershipV1(ownership ControlledSessionOwnershipV1) error { + if err := ValidateLiveRunIDV1(ownership.LiveRunID); err != nil { + return fmt.Errorf("live run ID: %w", err) + } + if err := validateBootSessionIDV1(ownership.BootSession); err != nil { + return err + } + if !controlledSessionHandlePatternV1.MatchString(ownership.SessionHandle) { + return fmt.Errorf("session handle must use session- followed by 64 lowercase hexadecimal characters") + } + 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") + } + if err := validateControlledSessionContainerOwnershipV1(ownership.Controller, "controller"); err != nil { + return fmt.Errorf("controller: %w", err) + } + if err := validateControlledSessionContainerOwnershipV1(ownership.Workload, "workload"); err != nil { + return fmt.Errorf("workload: %w", err) + } + if ownership.Controller.ID == ownership.Workload.ID { + return fmt.Errorf("controller and workload must name different containers") + } + return nil +} + +func validateControlledSessionContainerOwnershipV1(ownership ControlledSessionContainerOwnershipV1, role string) error { + if ownership.Role != role { + return fmt.Errorf("role must be %q", role) + } + if !controlledSessionContainerIDPatternV1.MatchString(ownership.ID) { + return fmt.Errorf("container ID must use 64 lowercase hexadecimal characters") + } + for label, value := range map[string]string{ + "name": ownership.Name, "deployment ID": ownership.DeploymentID, + "generation reference": ownership.GenerationReference, + } { + if !safeRecoveryIdentity(value) { + return fmt.Errorf("%s must be nonempty safe text", label) + } + } + if !controlledSessionBuildIdentityPatternV1.MatchString(ownership.BuildIdentity) { + return fmt.Errorf("build identity must be a sha256 digest") + } + return nil +} + func validateLiveRunContainerCleanupV1(cleanup LiveRunContainerCleanupV1) error { if !safeRecoveryIdentity(cleanup.Container) { return fmt.Errorf("cleanup container must be nonempty safe text") @@ -506,14 +583,19 @@ func ControlMarkersV1(queue LiveRunQueueV1) []ControlMarkerV1 { } func cloneLiveRunQueueV1(queue LiveRunQueueV1) LiveRunQueueV1 { + var controlledSessions []ControlledSessionOwnershipV1 + if queue.ControlledSessions != nil { + controlledSessions = append([]ControlledSessionOwnershipV1{}, queue.ControlledSessions...) + } var cleanup []LiveRunContainerCleanupV1 if queue.Cleanup != nil { cleanup = append([]LiveRunContainerCleanupV1{}, queue.Cleanup...) } return LiveRunQueueV1{ - Schema: queue.Schema, - Runs: append([]LiveRunV1{}, queue.Runs...), - Cleanup: cleanup, + Schema: queue.Schema, + Runs: append([]LiveRunV1{}, queue.Runs...), + ControlledSessions: controlledSessions, + Cleanup: cleanup, } } diff --git a/internal/deploy/live_run_queue_file.go b/internal/deploy/live_run_queue_file.go index ee11810e..a8635389 100644 --- a/internal/deploy/live_run_queue_file.go +++ b/internal/deploy/live_run_queue_file.go @@ -128,6 +128,108 @@ func (lock *OperationLock) RecordLiveRunContainerV1(id string, container string) return fmt.Errorf("live run %q is not outstanding", id) } +// RecordControlledSessionOwnershipV1 durably binds the exact inert resources +// to an active admitted shell before either controlled-session process starts. +// The boot identity comes from the admitted run already protected by this lock. +func (lock *OperationLock) RecordControlledSessionOwnershipV1(ownership ControlledSessionOwnershipV1) (ControlledSessionOwnershipV1, error) { + if lock == nil { + return ControlledSessionOwnershipV1{}, fmt.Errorf("record controlled session ownership requires an operation lock") + } + if err := ValidateLiveRunIDV1(ownership.LiveRunID); err != nil { + return ControlledSessionOwnershipV1{}, err + } + lock.mutex.Lock() + defer lock.mutex.Unlock() + path, err := lock.liveRunQueuePathLockedV1() + if err != nil { + return ControlledSessionOwnershipV1{}, err + } + queue, _, err := readLiveRunQueuePathV1(path) + if err != nil { + return ControlledSessionOwnershipV1{}, err + } + var admitted *LiveRunV1 + for index := range queue.Runs { + if queue.Runs[index].ID == ownership.LiveRunID { + admitted = &queue.Runs[index] + break + } + } + if admitted == nil { + return ControlledSessionOwnershipV1{}, fmt.Errorf("live run %q is not outstanding", ownership.LiveRunID) + } + if admitted.Status != LiveRunStatusActiveV1 || admitted.Kind != LiveRunKindShellV1 { + return ControlledSessionOwnershipV1{}, fmt.Errorf("controlled session live run %q must be an active shell", ownership.LiveRunID) + } + 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 := validateControlledSessionOwnershipV1(ownership); err != nil { + return ControlledSessionOwnershipV1{}, err + } + insert := sort.Search(len(queue.ControlledSessions), func(index int) bool { + return queue.ControlledSessions[index].LiveRunID >= ownership.LiveRunID + }) + if insert < len(queue.ControlledSessions) && queue.ControlledSessions[insert].LiveRunID == ownership.LiveRunID { + if queue.ControlledSessions[insert] != ownership { + return ControlledSessionOwnershipV1{}, fmt.Errorf("live run %q already has different controlled-session ownership", ownership.LiveRunID) + } + return ownership, nil + } + queue.ControlledSessions = append(queue.ControlledSessions, ControlledSessionOwnershipV1{}) + copy(queue.ControlledSessions[insert+1:], queue.ControlledSessions[insert:]) + queue.ControlledSessions[insert] = ownership + if err := commitLiveRunQueuePathV1(path, queue); err != nil { + return ControlledSessionOwnershipV1{}, err + } + return ownership, nil +} + +// CompleteControlledSessionV1 atomically removes a verified-clean session's +// ownership record and admitted run. Failed cleanup must not call this method. +func (lock *OperationLock) CompleteControlledSessionV1(id string) (bool, error) { + if lock == nil { + return false, fmt.Errorf("complete controlled session requires an operation lock") + } + if err := ValidateLiveRunIDV1(id); err != nil { + return false, err + } + lock.mutex.Lock() + defer lock.mutex.Unlock() + path, err := lock.liveRunQueuePathLockedV1() + if err != nil { + return false, err + } + queue, _, err := readLiveRunQueuePathV1(path) + if err != nil { + return false, err + } + updated, runRemoved, err := RemoveLiveRunV1(queue, id) + if err != nil { + return false, err + } + ownershipRemoved := false + for index, ownership := range updated.ControlledSessions { + if ownership.LiveRunID != id { + continue + } + updated.ControlledSessions = append(updated.ControlledSessions[:index], updated.ControlledSessions[index+1:]...) + ownershipRemoved = true + break + } + if !runRemoved && !ownershipRemoved { + return false, nil + } + if err := commitLiveRunQueuePathV1(path, updated); err != nil { + return false, err + } + return true, nil +} + func (lock *OperationLock) RemoveLiveRunV1(id string) (LiveRunQueueV1, bool, error) { if lock == nil { return LiveRunQueueV1{}, false, fmt.Errorf("remove live run requires an operation lock") @@ -518,7 +620,7 @@ func commitLiveRunQueuePathV1(path string, queue LiveRunQueueV1) error { if err != nil { return err } - if len(queue.Runs) == 0 && len(queue.Cleanup) == 0 { + if len(queue.Runs) == 0 && len(queue.ControlledSessions) == 0 && len(queue.Cleanup) == 0 { return removeLiveRunQueuePathV1(path) } if err := writeAtomicStateFile(path, content, 0o600); err != nil { diff --git a/internal/deploy/live_run_queue_file_test.go b/internal/deploy/live_run_queue_file_test.go index 1b4a7590..971e1979 100644 --- a/internal/deploy/live_run_queue_file_test.go +++ b/internal/deploy/live_run_queue_file_test.go @@ -60,6 +60,126 @@ func TestOperationLockLiveRunQueueFileLifecycle(t *testing.T) { } } +func TestOperationLockRecordsExactControlledSessionOwnership(t *testing.T) { + dir := t.TempDir() + lock, err := AcquireOperationLock(t.Context(), dir) + if err != nil { + t.Fatal(err) + } + defer lock.Unlock() + const runID = "run-0000000000000001" + const generation = "reploy/env/workload:g-current" + status, err := lock.AdmitLiveRunV1(LiveRunV1{ + ID: runID, Kind: LiveRunKindShellV1, Name: "controlled-session", + GenerationReference: generation, Exclusive: true, + }, false) + if err != nil || status != LiveRunStatusActiveV1 { + t.Fatalf("admission = %q, %v", status, err) + } + ownership := controlledSessionOwnershipFixtureV1(dir, runID, generation) + recorded, err := lock.RecordControlledSessionOwnershipV1(ownership) + if err != nil { + t.Fatal(err) + } + if recorded.BootSession == "" || ownership.BootSession != "" { + t.Fatalf("boot identity = recorded %q, input %q", recorded.BootSession, ownership.BootSession) + } + loaded, found, err := lock.ReadLiveRunQueueV1() + if err != nil || !found || len(loaded.ControlledSessions) != 1 || loaded.ControlledSessions[0] != recorded { + t.Fatalf("controlled-session ownership = %#v, found=%t, error=%v", loaded.ControlledSessions, found, err) + } + conflict := ownership + conflict.Controller.ID = strings.Repeat("c", 64) + if _, err := lock.RecordControlledSessionOwnershipV1(conflict); err == nil || !strings.Contains(err.Error(), "different controlled-session ownership") { + t.Fatalf("conflicting ownership error = %v", err) + } + if completed, err := lock.CompleteControlledSessionV1(runID); err != nil || !completed { + t.Fatalf("completion = %t, %v", completed, err) + } + if _, found, err := lock.ReadLiveRunQueueV1(); err != nil || found { + t.Fatalf("completed queue found=%t, error=%v", found, err) + } +} + +func TestOperationLockControlledSessionOwnershipWriteFailurePreservesQueue(t *testing.T) { + dir := t.TempDir() + lock, err := AcquireOperationLock(t.Context(), dir) + if err != nil { + t.Fatal(err) + } + defer lock.Unlock() + const runID = "run-0000000000000001" + const generation = "reploy/env/workload:g-current" + if status, err := lock.AdmitLiveRunV1(LiveRunV1{ + ID: runID, Kind: LiveRunKindShellV1, Name: "controlled-session", + GenerationReference: generation, Exclusive: true, + }, false); err != nil || status != LiveRunStatusActiveV1 { + t.Fatalf("admission = %q, %v", status, err) + } + before, _, err := lock.ReadLiveRunQueueV1() + if err != nil { + t.Fatal(err) + } + originalReplace := replaceAtomicStateFile + replaceAtomicStateFile = func(string, string) error { return errors.New("injected ownership replace failure") } + t.Cleanup(func() { replaceAtomicStateFile = originalReplace }) + if _, err := lock.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFixtureV1(dir, runID, generation)); err == nil || !strings.Contains(err.Error(), "injected ownership replace failure") { + t.Fatalf("ownership write error = %v", err) + } + after, _, err := lock.ReadLiveRunQueueV1() + if err != nil || !reflect.DeepEqual(after, before) { + t.Fatalf("failed ownership write changed queue: %#v, error=%v", after, err) + } +} + +func TestRecoverLiveRunQueuePreservesControlledSessionOwnership(t *testing.T) { + dir := t.TempDir() + lock, err := AcquireOperationLock(t.Context(), dir) + if err != nil { + t.Fatal(err) + } + defer lock.Unlock() + const runID = "run-0000000000000001" + const generation = "reploy/env/workload:g-current" + if status, err := lock.AdmitLiveRunV1(LiveRunV1{ + ID: runID, Kind: LiveRunKindShellV1, Name: "controlled-session", + GenerationReference: generation, Exclusive: true, + }, false); err != nil || status != LiveRunStatusActiveV1 { + t.Fatalf("admission = %q, %v", status, err) + } + recorded, err := lock.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFixtureV1(dir, runID, generation)) + if err != nil { + t.Fatal(err) + } + recovery, err := lock.RecoverLiveRunQueueV1() + if err != nil { + t.Fatal(err) + } + if len(recovery.Removed) != 1 || recovery.Removed[0].Run.ID != runID { + t.Fatalf("recovery = %#v", recovery) + } + queue, found, err := lock.ReadLiveRunQueueV1() + if err != nil || !found || len(queue.Runs) != 0 || len(queue.ControlledSessions) != 1 || queue.ControlledSessions[0] != recorded { + t.Fatalf("ownership after run recovery = %#v, found=%t, error=%v", queue, found, err) + } +} + +func controlledSessionOwnershipFixtureV1(dir string, runID string, generation string) ControlledSessionOwnershipV1 { + container := func(role string, id string, environment string, generation string, build string) ControlledSessionContainerOwnershipV1 { + return ControlledSessionContainerOwnershipV1{ + Role: role, ID: id, Name: "reploy-" + role + "-" + runID, + DeploymentID: environment, GenerationReference: generation, + BuildIdentity: "sha256:" + strings.Repeat(build, 64), + } + } + return ControlledSessionOwnershipV1{ + LiveRunID: runID, SessionHandle: "session-" + strings.Repeat("a", 64), + ChannelDirectory: filepath.Join(dir, ".reploy", "private", "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"), + } +} + func TestOperationLockLiveRunQueueReplaceFailurePreservesQueue(t *testing.T) { dir := t.TempDir() lock, err := AcquireOperationLock(t.Context(), dir) diff --git a/internal/dockerdeploy/controlled_session_controller.go b/internal/dockerdeploy/controlled_session_controller.go index 1d6fb17c..7f300338 100644 --- a/internal/dockerdeploy/controlled_session_controller.go +++ b/internal/dockerdeploy/controlled_session_controller.go @@ -40,6 +40,10 @@ type DockerControllerV1 struct { waitResult dockerControllerWaitResultV1 } +func (controller *DockerControllerV1) ContainerID() string { + return controller.containerID +} + // PrepareDockerControllerV1 verifies that the private channel is ready and // creates the exact controller container without starting it. func PrepareDockerControllerV1( diff --git a/internal/dockerdeploy/controlled_session_supervisor.go b/internal/dockerdeploy/controlled_session_supervisor.go index 1daa4c73..756a4aa3 100644 --- a/internal/dockerdeploy/controlled_session_supervisor.go +++ b/internal/dockerdeploy/controlled_session_supervisor.go @@ -5,10 +5,12 @@ import ( "errors" "fmt" "io" + "path/filepath" "sync" "time" "github.com/omry/reploy/internal/controlledsession" + "github.com/omry/reploy/internal/deploy" ) const controlledSessionOutputFinalizationTimeoutV1 = time.Duration(controlledsession.DefaultOutputFinalizationTimeoutMillisecondsV1) * time.Millisecond @@ -34,6 +36,7 @@ type ControlledSessionRunResultV1 struct { } type controlledSessionControllerRuntimeV1 interface { + ContainerID() string Start(context.Context) error Wait(context.Context) (controlledsession.ProcessStatusV1, error) RequestGracefulStop(context.Context) error @@ -43,6 +46,7 @@ type controlledSessionControllerRuntimeV1 interface { type controlledSessionWorkloadRuntimeV1 interface { controlledsession.WorkloadPTYControlV1 + ContainerID() string Output() (io.ReadCloser, error) Start(context.Context) error Started() bool @@ -74,6 +78,7 @@ type controlledSessionSupervisorBackendV1 struct { prepareChannel func(ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) prepareController func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionControllerRuntimeV1, error) prepareWorkload func(context.Context, ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) + recordOwnership func(string, string) error now func() time.Time } @@ -120,16 +125,39 @@ type controlledSessionSupervisorV1 struct { diagnosticErr error } -// RunControlledSessionV1 owns one attached controller/workload operation from -// inert resource creation through terminal acknowledgement and ordinary -// delivery-tail cleanup. Crash reconciliation, watchdog ownership, networking, -// and public command exposure are deliberately outside this lifecycle core. +// RunControlledSessionV1 takes ownership of the admitted workload operation +// lock. It durably records both exact inert containers and the private channel +// before releasing the lock and starting either process. func RunControlledSessionV1( ctx context.Context, + operation *deploy.OperationLock, plan ControlledSessionExecutionPlanV1, options ControlledSessionRunOptionsV1, ) (ControlledSessionRunResultV1, error) { - return runControlledSessionV1(ctx, plan, options, controlledSessionSupervisorBackendV1{ + if operation == nil { + return ControlledSessionRunResultV1{}, fmt.Errorf("run controlled session requires an admitted operation lock") + } + if err := operation.RequireHeld(); err != nil { + return ControlledSessionRunResultV1{}, err + } + absoluteDir, err := filepath.Abs(plan.Workload.DeploymentDirectory) + if err != nil { + return ControlledSessionRunResultV1{}, releaseControlledSessionOperationV1(operation, 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)) + } + if ctx == nil || ctx.Done() == nil { + return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionV1(operation, 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)) + } + if err := validateControlledSessionRunOptionsV1(options); err != nil { + return ControlledSessionRunResultV1{}, removeUnstartedControlledSessionV1(operation, plan.LiveRunID, err) + } + released := false + result, runErr := runControlledSessionV1(ctx, plan, options, controlledSessionSupervisorBackendV1{ prepareChannel: func(plan ControlledSessionExecutionPlanV1) (controlledSessionChannelRuntimeV1, error) { channel, err := PrepareControlledSessionChannelV1(plan) if err != nil { @@ -143,8 +171,81 @@ func RunControlledSessionV1( prepareWorkload: func(ctx context.Context, plan ControlledSessionContainerPlanV1) (controlledSessionWorkloadRuntimeV1, error) { return PrepareDockerWorkloadPTYV1(ctx, plan) }, + recordOwnership: func(controllerID string, workloadID string) error { + ownership := controlledSessionOwnershipFromPlanV1(plan, controllerID, workloadID) + if _, err := operation.RecordControlledSessionOwnershipV1(ownership); err != nil { + return fmt.Errorf("persist controlled-session ownership: %w", err) + } + released = true + if err := operation.Unlock(); err != nil { + return fmt.Errorf("release operation lock before controlled-session startup: %w", err) + } + return nil + }, now: time.Now, }) + cleaned := result.SessionResult.CleanupStatus.Kind == controlledsession.CleanupStatusSucceededV1 && + result.DeliveryTailCleanupStatus.Kind == controlledsession.CleanupStatusSucceededV1 + completionErr := finishControlledSessionOwnershipV1(context.WithoutCancel(ctx), absoluteDir, operation, released, plan.LiveRunID, cleaned) + return result, errors.Join(runErr, completionErr) +} + +func controlledSessionOwnershipFromPlanV1(plan ControlledSessionExecutionPlanV1, 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, + GenerationReference: plan.GenerationReference, BuildIdentity: string(plan.BuildIdentity), + } + } + return deploy.ControlledSessionOwnershipV1{ + LiveRunID: plan.LiveRunID, SessionHandle: plan.Authorization.Handle, + ChannelDirectory: plan.Channel.HostDirectory, + Controller: container(plan.Controller, controllerID), Workload: container(plan.Workload, workloadID), + } +} + +func finishControlledSessionOwnershipV1( + ctx context.Context, + deploymentDir string, + operation *deploy.OperationLock, + released bool, + runID string, + cleaned bool, +) error { + if released { + var err error + operation, err = deploy.AcquireOperationLock(ctx, deploymentDir) + if err != nil { + return fmt.Errorf("reacquire operation lock after controlled session: %w", err) + } + } + var completionErr error + if cleaned { + _, completionErr = operation.CompleteControlledSessionV1(runID) + if completionErr != nil { + completionErr = fmt.Errorf("remove verified-clean controlled-session ownership: %w", completionErr) + } + } + unlockErr := operation.Unlock() + if unlockErr != nil { + unlockErr = fmt.Errorf("release controlled-session operation lock: %w", unlockErr) + } + return errors.Join(completionErr, unlockErr) +} + +func removeUnstartedControlledSessionV1(operation *deploy.OperationLock, runID string, cause error) error { + var removeErr error + if deploy.ValidateLiveRunIDV1(runID) == nil { + _, _, removeErr = operation.RemoveLiveRunV1(runID) + } + return releaseControlledSessionOperationV1(operation, errors.Join(cause, removeErr)) +} + +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)) + } + return cause } func runControlledSessionV1( @@ -262,6 +363,11 @@ func (supervisor *controlledSessionSupervisorV1) prepare(ctx context.Context) er if err != nil { return fmt.Errorf("claim controlled-session workload output: %w", err) } + if supervisor.backend.recordOwnership != nil { + if err := supervisor.backend.recordOwnership(controller.ContainerID(), workload.ContainerID()); err != nil { + return err + } + } if err := controller.Start(ctx); err != nil { return fmt.Errorf("start controlled-session controller: %w", err) } diff --git a/internal/dockerdeploy/controlled_session_supervisor_integration_test.go b/internal/dockerdeploy/controlled_session_supervisor_integration_test.go index 7240a392..60b01f05 100644 --- a/internal/dockerdeploy/controlled_session_supervisor_integration_test.go +++ b/internal/dockerdeploy/controlled_session_supervisor_integration_test.go @@ -10,6 +10,7 @@ import ( "time" "github.com/omry/reploy/internal/controlledsession" + "github.com/omry/reploy/internal/deploy" ) func TestControlledSessionSupervisorDockerIntegration(t *testing.T) { @@ -23,8 +24,23 @@ func TestControlledSessionSupervisorDockerIntegration(t *testing.T) { defer cancel() image := buildControlledSessionControllerIntegrationImageV1(t, ctx) plan := controlledSessionControllerIntegrationPlanV1(t, image, []string{"/session-channel-helper", "supervise"}) + operation, err := deploy.AcquireOperationLock(ctx, plan.Workload.DeploymentDirectory) + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = operation.Unlock() }) + status, err := operation.AdmitLiveRunV1(deploy.LiveRunV1{ + ID: plan.LiveRunID, Kind: deploy.LiveRunKindShellV1, Name: plan.Workload.DeploymentID, + GenerationReference: plan.Workload.GenerationReference, Exclusive: true, + }, false) + if err != nil { + t.Fatal(err) + } + if status != deploy.LiveRunStatusActiveV1 { + t.Fatalf("controlled-session live run status = %q", status) + } - result, err := RunControlledSessionV1(ctx, plan, ControlledSessionRunOptionsV1{ + result, err := RunControlledSessionV1(ctx, operation, plan, ControlledSessionRunOptionsV1{ StartupTimeout: 30 * time.Second, TerminationGrace: 5 * time.Second, ControllerFinalizationTimeout: 15 * time.Second, ResultAcknowledgementTimeout: 5 * time.Second, CleanupTimeout: 15 * time.Second, @@ -53,4 +69,12 @@ func TestControlledSessionSupervisorDockerIntegration(t *testing.T) { if _, statErr := os.Stat(plan.Channel.HostDirectory); !os.IsNotExist(statErr) { t.Fatalf("private channel directory survived cleanup: %v", statErr) } + check, lockErr := deploy.AcquireOperationLock(ctx, plan.Workload.DeploymentDirectory) + if lockErr != nil { + t.Fatal(lockErr) + } + defer check.Unlock() + if queue, found, readErr := check.ReadLiveRunQueueV1(); readErr != nil || found { + t.Fatalf("verified-clean session retained ownership: %#v, found=%t, error=%v", queue, found, readErr) + } } diff --git a/internal/dockerdeploy/controlled_session_supervisor_test.go b/internal/dockerdeploy/controlled_session_supervisor_test.go index 67443e49..84b61781 100644 --- a/internal/dockerdeploy/controlled_session_supervisor_test.go +++ b/internal/dockerdeploy/controlled_session_supervisor_test.go @@ -4,11 +4,13 @@ import ( "context" "errors" "io" + "strings" "sync" "testing" "time" "github.com/omry/reploy/internal/controlledsession" + "github.com/omry/reploy/internal/deploy" ) func TestRunControlledSessionV1OwnsNormalLifecycle(t *testing.T) { @@ -91,6 +93,109 @@ func TestRunControlledSessionV1OwnsNormalLifecycle(t *testing.T) { } } +func TestRunControlledSessionV1PersistsExactOwnershipBeforeStarting(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + controller := newFakeControlledSessionProcessV1() + workload := newFakeControlledSessionWorkloadV1(nil, 0) + channel := &fakeControlledSessionChannelV1{} + persistErr := errors.New("injected durable ownership failure") + called := false + + 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 + }, + recordOwnership: func(controllerID string, workloadID string) error { + called = true + if controller.started || workload.started { + t.Fatal("controlled-session process started before durable ownership") + } + if controllerID != dockerControllerTestContainerIDV1 || workloadID != dockerWorkloadTestContainerIDV1 { + t.Fatalf("container IDs = %q / %q", controllerID, workloadID) + } + return persistErr + }, + now: time.Now, + }) + if !called || !errors.Is(err, persistErr) { + t.Fatalf("ownership persistence called=%t, error=%v", called, err) + } + if controller.started || workload.started { + t.Fatalf("started after persistence failure = controller %t workload %t", controller.started, workload.started) + } + if !controller.cleaned || !workload.cleaned || !channel.closed { + t.Fatalf("inert cleanup = controller %t workload %t channel %t", controller.cleaned, workload.cleaned, channel.closed) + } + if result.SessionResult.CleanupStatus.Kind != controlledsession.CleanupStatusSucceededV1 || + result.DeliveryTailCleanupStatus.Kind != controlledsession.CleanupStatusSucceededV1 { + t.Fatalf("cleanup result = %#v", result) + } +} + +func TestControlledSessionCleanupFailureRetainsDurableOwnership(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) + } + if status, err := operation.AdmitLiveRunV1(deploy.LiveRunV1{ + ID: plan.LiveRunID, Kind: deploy.LiveRunKindShellV1, Name: plan.Workload.DeploymentID, + GenerationReference: plan.Workload.GenerationReference, Exclusive: true, + }, false); err != nil || status != deploy.LiveRunStatusActiveV1 { + t.Fatalf("admission = %q, %v", status, err) + } + if _, err := operation.RecordControlledSessionOwnershipV1(controlledSessionOwnershipFromPlanV1( + plan, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1, + )); err != nil { + t.Fatal(err) + } + if err := finishControlledSessionOwnershipV1(t.Context(), plan.Workload.DeploymentDirectory, operation, false, plan.LiveRunID, false); err != nil { + t.Fatal(err) + } + check, err := deploy.AcquireOperationLock(t.Context(), plan.Workload.DeploymentDirectory) + if err != nil { + t.Fatal(err) + } + defer check.Unlock() + queue, found, err := check.ReadLiveRunQueueV1() + if err != nil || !found || len(queue.Runs) != 1 || len(queue.ControlledSessions) != 1 { + t.Fatalf("retained queue = %#v, found=%t, error=%v", queue, found, err) + } +} + +func TestRunControlledSessionV1RemovesAdmissionOnLockDirectoryMismatch(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + workloadDir := plan.Workload.DeploymentDirectory + operation, err := deploy.AcquireOperationLock(t.Context(), workloadDir) + if err != nil { + t.Fatal(err) + } + if status, err := operation.AdmitLiveRunV1(deploy.LiveRunV1{ + ID: plan.LiveRunID, Kind: deploy.LiveRunKindShellV1, Name: plan.Workload.DeploymentID, + GenerationReference: plan.Workload.GenerationReference, Exclusive: true, + }, false); err != nil || status != deploy.LiveRunStatusActiveV1 { + t.Fatalf("admission = %q, %v", status, err) + } + plan.Workload.DeploymentDirectory = t.TempDir() + if _, err := RunControlledSessionV1(t.Context(), operation, plan, testControlledSessionRunOptionsV1()); err == nil || !strings.Contains(err.Error(), "does not belong") { + t.Fatalf("lock-directory mismatch error = %v", err) + } + check, err := deploy.AcquireOperationLock(t.Context(), workloadDir) + if err != nil { + t.Fatal(err) + } + defer check.Unlock() + if queue, found, err := check.ReadLiveRunQueueV1(); err != nil || found { + t.Fatalf("mismatched operation retained admission: %#v, found=%t, error=%v", queue, found, err) + } +} + func TestRunControlledSessionV1HoldsCompleteUntilOutputFinalizationPublication(t *testing.T) { plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) requests := make(chan controlledsession.RequestV1, 8) @@ -786,6 +891,10 @@ type fakeControlledSessionProcessV1 struct { cleaned bool } +func (process *fakeControlledSessionProcessV1) ContainerID() string { + return dockerControllerTestContainerIDV1 +} + func newFakeControlledSessionProcessV1() *fakeControlledSessionProcessV1 { return &fakeControlledSessionProcessV1{exit: make(chan controlledSessionProcessResultV1, 1)} } @@ -843,6 +952,10 @@ type fakeControlledSessionWorkloadV1 struct { rows uint32 } +func (workload *fakeControlledSessionWorkloadV1) ContainerID() string { + return dockerWorkloadTestContainerIDV1 +} + func newFakeControlledSessionWorkloadV1(output []byte, exitCode int) *fakeControlledSessionWorkloadV1 { reader, writer := io.Pipe() return &fakeControlledSessionWorkloadV1{ diff --git a/internal/dockerdeploy/controlled_session_workload_pty.go b/internal/dockerdeploy/controlled_session_workload_pty.go index f0fea71a..37acc936 100644 --- a/internal/dockerdeploy/controlled_session_workload_pty.go +++ b/internal/dockerdeploy/controlled_session_workload_pty.go @@ -54,6 +54,10 @@ type DockerWorkloadPTYV1 struct { closeErr error } +func (workload *DockerWorkloadPTYV1) ContainerID() string { + return workload.containerID +} + // PrepareDockerWorkloadPTYV1 creates the exact workload container without // starting it and establishes the Docker PTY attachment before returning. func PrepareDockerWorkloadPTYV1(