From 57c5377ef00b8a0584b523e97e4c90815ccc7ccd Mon Sep 17 00:00:00 2001 From: Omry Yadan Date: Tue, 11 Aug 2026 02:08:36 +0800 Subject: [PATCH] Persist controlled-session network ownership Persist controlled-session network ownership alongside container and channel ownership. Carry the exact network through cleanup manifests, live-run state, watchdog receipts, and restart recovery so cleanup cannot omit or target the wrong network. --- .../controlled_session_cleanup_manifest.go | 25 ++- ...ontrolled_session_cleanup_manifest_test.go | 28 +++- .../controlled_session_incident_receipt.go | 31 ++++ ...ontrolled_session_incident_receipt_test.go | 28 ++++ internal/deploy/live_run_queue.go | 38 +++++ internal/deploy/live_run_queue_file.go | 20 ++- internal/deploy/live_run_queue_file_test.go | 61 +++++++ .../controlled_session_incident_test.go | 3 +- .../controlled_session_network.go | 15 +- .../controlled_session_supervisor_test.go | 4 +- .../controlled_session_watchdog.go | 131 ++++++++++++++- .../controlled_session_watchdog_test.go | 156 +++++++++++++++++- internal/dockerdeploy/live_run_recovery.go | 98 +++++++++++ .../dockerdeploy/live_run_recovery_test.go | 140 +++++++++++++++- 14 files changed, 747 insertions(+), 31 deletions(-) diff --git a/internal/deploy/controlled_session_cleanup_manifest.go b/internal/deploy/controlled_session_cleanup_manifest.go index 5f807091..8537a72c 100644 --- a/internal/deploy/controlled_session_cleanup_manifest.go +++ b/internal/deploy/controlled_session_cleanup_manifest.go @@ -14,9 +14,9 @@ import ( // selection a session watchdog may receive. It is derived from durable // ownership rather than accepting later resource choices. // -// Networks and volumes are explicit arrays even while controlled sessions do -// not create either resource. Future slices may populate them only after their -// exact identities become part of durable ownership. +// Networks and volumes are explicit arrays. Network entries carry the exact +// immutable identity selected by durable ownership; volumes remain reserved +// for a later resource slice. type ControlledSessionCleanupManifest struct { LiveRunID string `json:"live_run_id"` BootSession string `json:"boot_session"` @@ -25,7 +25,7 @@ type ControlledSessionCleanupManifest struct { IncidentReceipt string `json:"incident_receipt"` Controller ControlledSessionContainerOwnershipV1 `json:"controller"` Workload ControlledSessionContainerOwnershipV1 `json:"workload"` - Networks []string `json:"networks"` + Networks []ControlledSessionNetworkOwnershipV1 `json:"networks"` Volumes []string `json:"volumes"` } @@ -40,12 +40,18 @@ func ControlledSessionCleanupManifestFromOwnership(ownership ControlledSessionOw if err != nil { return ControlledSessionCleanupManifest{}, err } + networks := []ControlledSessionNetworkOwnershipV1{} + if ownership.NetworkName != "" { + networks = append(networks, ControlledSessionNetworkOwnershipV1{ + Role: ControlledSessionNetworkRoleV1, ID: ownership.NetworkID, Name: ownership.NetworkName, + }) + } manifest := ControlledSessionCleanupManifest{ LiveRunID: ownership.LiveRunID, BootSession: ownership.BootSession, DockerEndpoint: ownership.DockerEndpoint, ChannelDirectory: ownership.ChannelDirectory, IncidentReceipt: receiptPath, Controller: ownership.Controller, Workload: ownership.Workload, - Networks: []string{}, Volumes: []string{}, + Networks: networks, Volumes: []string{}, } if err := ValidateControlledSessionCleanupManifest(manifest); err != nil { return ControlledSessionCleanupManifest{}, err @@ -87,8 +93,13 @@ func ValidateControlledSessionCleanupManifest(manifest ControlledSessionCleanupM if manifest.Networks == nil || manifest.Volumes == nil { return fmt.Errorf("controlled-session cleanup manifest networks and volumes must use arrays") } - if err := validateControlledSessionCleanupResourceIDs(manifest.Networks, "network"); err != nil { - return err + if len(manifest.Networks) > 1 { + return fmt.Errorf("controlled-session cleanup manifest may name only one network") + } + for index, network := range manifest.Networks { + if err := validateControlledSessionNetworkOwnershipV1(network); err != nil { + return fmt.Errorf("controlled-session cleanup manifest network %d: %w", index, err) + } } if err := validateControlledSessionCleanupResourceIDs(manifest.Volumes, "volume"); err != nil { return err diff --git a/internal/deploy/controlled_session_cleanup_manifest_test.go b/internal/deploy/controlled_session_cleanup_manifest_test.go index 52de7848..8cd4b254 100644 --- a/internal/deploy/controlled_session_cleanup_manifest_test.go +++ b/internal/deploy/controlled_session_cleanup_manifest_test.go @@ -11,15 +11,23 @@ import ( func TestControlledSessionCleanupManifestDerivesExactDurableOwnership(t *testing.T) { ownership := controlledSessionOwnershipFixtureV1(t.TempDir(), "run-0000000000000001", "reploy/env/workload:g-current") ownership.BootSession = "boot-session" + ownership.NetworkID = strings.Repeat("d", 64) + ownership.NetworkName = "reploy-session-network" manifest, err := ControlledSessionCleanupManifestFromOwnership(ownership) if err != nil { t.Fatal(err) } if manifest.LiveRunID != ownership.LiveRunID || manifest.BootSession != ownership.BootSession || manifest.DockerEndpoint != ownership.DockerEndpoint || manifest.ChannelDirectory != ownership.ChannelDirectory || manifest.Controller != ownership.Controller || - manifest.Workload != ownership.Workload || len(manifest.Networks) != 0 || len(manifest.Volumes) != 0 { + manifest.Workload != ownership.Workload || len(manifest.Networks) != 1 || len(manifest.Volumes) != 0 { t.Fatalf("cleanup manifest = %#v", manifest) } + wantNetwork := ControlledSessionNetworkOwnershipV1{ + Role: ControlledSessionNetworkRoleV1, ID: ownership.NetworkID, Name: ownership.NetworkName, + } + if manifest.Networks[0] != wantNetwork { + t.Fatalf("cleanup manifest network = %#v, want %#v", manifest.Networks[0], wantNetwork) + } wantReceipt := filepath.Join(filepath.Dir(filepath.Dir(ownership.ChannelDirectory)), "incidents", ownership.LiveRunID+".json") if manifest.IncidentReceipt != wantReceipt { t.Fatalf("incident receipt = %q, want %q", manifest.IncidentReceipt, wantReceipt) @@ -37,6 +45,15 @@ func TestControlledSessionCleanupManifestDerivesExactDurableOwnership(t *testing } } +func TestControlledSessionCleanupManifestRejectsPlannedNetworkWithoutExactID(t *testing.T) { + ownership := controlledSessionOwnershipFixtureV1(t.TempDir(), "run-0000000000000001", "reploy/env/workload:g-current") + ownership.BootSession = "boot-session" + ownership.NetworkName = "reploy-session-network" + if _, err := ControlledSessionCleanupManifestFromOwnership(ownership); err == nil || !strings.Contains(err.Error(), "network ID") { + t.Fatalf("incomplete network ownership error = %v", err) + } +} + func TestControlledSessionCleanupManifestRejectsRemoteDockerEndpoint(t *testing.T) { ownership := controlledSessionOwnershipFixtureV1(t.TempDir(), "run-0000000000000001", "reploy/env/workload:g-current") ownership.BootSession = "boot-session" @@ -71,9 +88,12 @@ func TestControlledSessionCleanupManifestRequiresExactChannelAndCanonicalArrays( if err := ValidateControlledSessionCleanupManifest(manifest); err == nil || !strings.Contains(err.Error(), "must use arrays") { t.Fatalf("nil resources error = %v", err) } - manifest.Networks = []string{"network-b", "network-a"} - if err := ValidateControlledSessionCleanupManifest(manifest); err == nil || !strings.Contains(err.Error(), "sorted and unique") { - t.Fatalf("unordered resources error = %v", err) + manifest.Networks = []ControlledSessionNetworkOwnershipV1{ + {Role: ControlledSessionNetworkRoleV1, ID: strings.Repeat("a", 64), Name: "network-a"}, + {Role: ControlledSessionNetworkRoleV1, ID: strings.Repeat("b", 64), Name: "network-b"}, + } + if err := ValidateControlledSessionCleanupManifest(manifest); err == nil || !strings.Contains(err.Error(), "only one network") { + t.Fatalf("multiple network resources error = %v", err) } } diff --git a/internal/deploy/controlled_session_incident_receipt.go b/internal/deploy/controlled_session_incident_receipt.go index 092f1f02..e19fe6ff 100644 --- a/internal/deploy/controlled_session_incident_receipt.go +++ b/internal/deploy/controlled_session_incident_receipt.go @@ -60,6 +60,16 @@ type ControlledSessionIncidentContainerV1 struct { CleanupStatus ControlledSessionIncidentResourceStatusV1 `json:"cleanup_status"` } +// ControlledSessionIncidentNetworkV1 carries only the immutable network +// identity and its allowlisted cleanup outcome. The ownership labels are the +// receipt live-run ID and the fixed network role. +type ControlledSessionIncidentNetworkV1 struct { + Role string `json:"role"` + ID string `json:"id"` + Name string `json:"name"` + CleanupStatus ControlledSessionIncidentResourceStatusV1 `json:"cleanup_status"` +} + // ControlledSessionIncidentReceiptV1 is the bounded durable evidence written // by the session watchdog after loss of its parent. The fixed fields cannot // carry PTY bytes, environment values, secrets, arbitrary logs, or raw Docker @@ -72,6 +82,7 @@ type ControlledSessionIncidentReceiptV1 struct { Trigger ControlledSessionIncidentTriggerV1 `json:"trigger"` Controller ControlledSessionIncidentContainerV1 `json:"controller"` Workload ControlledSessionIncidentContainerV1 `json:"workload"` + Networks []ControlledSessionIncidentNetworkV1 `json:"networks,omitempty"` ChannelCleanupStatus ControlledSessionIncidentResourceStatusV1 `json:"channel_cleanup_status"` CleanupStatus ControlledSessionIncidentCleanupStatusV1 `json:"cleanup_status"` RecoveryAction ControlledSessionIncidentRecoveryActionV1 `json:"recovery_action"` @@ -346,12 +357,32 @@ func ValidateControlledSessionIncidentReceiptV1(receipt ControlledSessionInciden if receipt.Controller.ID == receipt.Workload.ID { return fmt.Errorf("controlled-session incident receipt containers must be different") } + if len(receipt.Networks) > 1 { + return fmt.Errorf("controlled-session incident receipt may name only one network") + } + for index, network := range receipt.Networks { + if network.Role != ControlledSessionNetworkRoleV1 { + return fmt.Errorf("controlled-session incident receipt network %d role must be %q", index, ControlledSessionNetworkRoleV1) + } + if !controlledSessionContainerIDPatternV1.MatchString(network.ID) { + return fmt.Errorf("controlled-session incident receipt network %d ID must use 64 lowercase hexadecimal characters", index) + } + if !safeRecoveryIdentity(network.Name) { + return fmt.Errorf("controlled-session incident receipt network %d name must be nonempty safe text", index) + } + if err := validateControlledSessionIncidentResourceStatusV1(network.CleanupStatus); err != nil { + return fmt.Errorf("controlled-session incident receipt network %d: %w", index, err) + } + } if err := validateControlledSessionIncidentResourceStatusV1(receipt.ChannelCleanupStatus); err != nil { return fmt.Errorf("controlled-session incident receipt channel: %w", err) } allSucceeded := receipt.Controller.CleanupStatus == ControlledSessionIncidentResourceVerifiedAbsentV1 && receipt.Workload.CleanupStatus == ControlledSessionIncidentResourceVerifiedAbsentV1 && receipt.ChannelCleanupStatus == ControlledSessionIncidentResourceVerifiedAbsentV1 + for _, network := range receipt.Networks { + allSucceeded = allSucceeded && network.CleanupStatus == ControlledSessionIncidentResourceVerifiedAbsentV1 + } switch receipt.CleanupStatus { case ControlledSessionIncidentCleanupSucceededV1: if !allSucceeded || receipt.RecoveryAction != ControlledSessionIncidentRecoveryNoneV1 { diff --git a/internal/deploy/controlled_session_incident_receipt_test.go b/internal/deploy/controlled_session_incident_receipt_test.go index dabc83d7..ee858587 100644 --- a/internal/deploy/controlled_session_incident_receipt_test.go +++ b/internal/deploy/controlled_session_incident_receipt_test.go @@ -37,6 +37,30 @@ func TestControlledSessionIncidentReceiptRoundTripHasOnlyAllowlistedFacts(t *tes } } +func TestControlledSessionIncidentReceiptValidatesExactNetworkIdentityAndOutcome(t *testing.T) { + for _, test := range []struct { + name string + mutate func(*ControlledSessionIncidentReceiptV1) + want string + }{ + {name: "role", mutate: func(value *ControlledSessionIncidentReceiptV1) { value.Networks[0].Role = "other" }, want: "role"}, + {name: "ID", mutate: func(value *ControlledSessionIncidentReceiptV1) { value.Networks[0].ID = "short" }, want: "64 lowercase"}, + {name: "name", mutate: func(value *ControlledSessionIncidentReceiptV1) { value.Networks[0].Name = "" }, want: "name"}, + {name: "outcome", mutate: func(value *ControlledSessionIncidentReceiptV1) { value.Networks[0].CleanupStatus = "unknown" }, want: "cleanup status"}, + {name: "overall success", mutate: func(value *ControlledSessionIncidentReceiptV1) { + value.Networks[0].CleanupStatus = ControlledSessionIncidentResourceCleanupFailedV1 + }, want: "every resource verified absent"}, + } { + t.Run(test.name, func(t *testing.T) { + receipt := controlledSessionIncidentReceiptFixtureV1("run-0000000000000001") + test.mutate(&receipt) + if err := ValidateControlledSessionIncidentReceiptV1(receipt); err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("network receipt validation error = %v", err) + } + }) + } +} + func TestOperationLockPreparesRetrievesAndAcknowledgesExactIncidentReceipt(t *testing.T) { dir := t.TempDir() lock, err := AcquireOperationLock(t.Context(), dir) @@ -226,6 +250,10 @@ func controlledSessionIncidentReceiptFixtureV1(runID string) ControlledSessionIn RecordedAt: time.Date(2026, 8, 10, 3, 0, 0, 0, time.UTC).Format(time.RFC3339Nano), Trigger: ControlledSessionIncidentParentLostV1, Controller: container("controller", "c"), Workload: container("workload", "d"), + Networks: []ControlledSessionIncidentNetworkV1{{ + Role: ControlledSessionNetworkRoleV1, ID: strings.Repeat("e", 64), Name: "reploy-session-network", + CleanupStatus: ControlledSessionIncidentResourceVerifiedAbsentV1, + }}, ChannelCleanupStatus: ControlledSessionIncidentResourceVerifiedAbsentV1, CleanupStatus: ControlledSessionIncidentCleanupSucceededV1, RecoveryAction: ControlledSessionIncidentRecoveryNoneV1, diff --git a/internal/deploy/live_run_queue.go b/internal/deploy/live_run_queue.go index 34d2cfc6..0a75a79a 100644 --- a/internal/deploy/live_run_queue.go +++ b/internal/deploy/live_run_queue.go @@ -69,10 +69,20 @@ type ControlledSessionOwnershipV1 struct { 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"` } +const ControlledSessionNetworkRoleV1 = "network" + +type ControlledSessionNetworkOwnershipV1 struct { + Role string `json:"role"` + ID string `json:"id"` + Name string `json:"name"` +} + type ControlledSessionContainerOwnershipV1 struct { Role string `json:"role"` ID string `json:"id"` @@ -299,6 +309,18 @@ func validateControlledSessionOwnershipV1(ownership ControlledSessionOwnershipV1 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 ownership.NetworkName == "" { + if ownership.NetworkID != "" { + return fmt.Errorf("network ID cannot be recorded without a network name") + } + } else { + if !safeRecoveryIdentity(ownership.NetworkName) { + return fmt.Errorf("network name must be nonempty safe text") + } + if ownership.NetworkID != "" && !controlledSessionContainerIDPatternV1.MatchString(ownership.NetworkID) { + return fmt.Errorf("network ID must use 64 lowercase hexadecimal characters") + } + } if err := validateControlledSessionContainerOwnershipStateV1(ownership.Controller, "controller"); err != nil { return fmt.Errorf("controller: %w", err) } @@ -308,12 +330,28 @@ func validateControlledSessionOwnershipV1(ownership ControlledSessionOwnershipV1 if ownership.Controller.ID == "" && ownership.Workload.ID != "" { return fmt.Errorf("workload container ID cannot be recorded before the controller container ID") } + if ownership.NetworkName != "" && ownership.NetworkID == "" && (ownership.Controller.ID != "" || ownership.Workload.ID != "") { + return fmt.Errorf("container IDs cannot be recorded before the network ID") + } if ownership.Controller.ID != "" && ownership.Workload.ID != "" && ownership.Controller.ID == ownership.Workload.ID { return fmt.Errorf("controller and workload must name different containers") } return nil } +func validateControlledSessionNetworkOwnershipV1(ownership ControlledSessionNetworkOwnershipV1) error { + if ownership.Role != ControlledSessionNetworkRoleV1 { + return fmt.Errorf("role must be %q", ControlledSessionNetworkRoleV1) + } + if !controlledSessionContainerIDPatternV1.MatchString(ownership.ID) { + return fmt.Errorf("network ID must use 64 lowercase hexadecimal characters") + } + if !safeRecoveryIdentity(ownership.Name) { + return fmt.Errorf("network name must be nonempty safe text") + } + return nil +} + func validateCurrentControlledSessionOwnershipV1(ownership ControlledSessionOwnershipV1) error { if ownership.DockerEndpoint == "" { return fmt.Errorf("Docker endpoint must be recorded for a new controlled session") diff --git a/internal/deploy/live_run_queue_file.go b/internal/deploy/live_run_queue_file.go index 72341233..aaa6317f 100644 --- a/internal/deploy/live_run_queue_file.go +++ b/internal/deploy/live_run_queue_file.go @@ -129,9 +129,9 @@ func (lock *OperationLock) RecordLiveRunContainerV1(id string, container string) } // RecordControlledSessionOwnershipV1 durably binds the planned resources to an -// active admitted shell and monotonically fills each exact container ID after -// Docker returns it. The boot identity comes from the admitted run already -// protected by this lock. +// active admitted shell and monotonically fills each exact network and +// container ID after Docker returns it. 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") @@ -204,6 +204,8 @@ func mergeControlledSessionOwnershipV1( ) (ControlledSessionOwnershipV1, error) { existingPlan := existing requestedPlan := requested + existingPlan.NetworkID = "" + requestedPlan.NetworkID = "" existingPlan.Controller.ID = "" existingPlan.Workload.ID = "" requestedPlan.Controller.ID = "" @@ -212,21 +214,25 @@ func mergeControlledSessionOwnershipV1( return ControlledSessionOwnershipV1{}, fmt.Errorf("immutable resource plan changed") } merged := existing - mergeID := func(current string, next string, role string) (string, error) { + mergeID := func(current string, next string, resource string) (string, error) { if next == "" { return current, nil } if current != "" && current != next { - return "", fmt.Errorf("%s container ID changed", role) + return "", fmt.Errorf("%s ID changed", resource) } return next, nil } var err error - merged.Controller.ID, err = mergeID(existing.Controller.ID, requested.Controller.ID, "controller") + merged.NetworkID, err = mergeID(existing.NetworkID, requested.NetworkID, "network") if err != nil { return ControlledSessionOwnershipV1{}, err } - merged.Workload.ID, err = mergeID(existing.Workload.ID, requested.Workload.ID, "workload") + merged.Controller.ID, err = mergeID(existing.Controller.ID, requested.Controller.ID, "controller container") + if err != nil { + return ControlledSessionOwnershipV1{}, err + } + merged.Workload.ID, err = mergeID(existing.Workload.ID, requested.Workload.ID, "workload container") if err != nil { return ControlledSessionOwnershipV1{}, err } diff --git a/internal/deploy/live_run_queue_file_test.go b/internal/deploy/live_run_queue_file_test.go index 0257d266..1cae9b7c 100644 --- a/internal/deploy/live_run_queue_file_test.go +++ b/internal/deploy/live_run_queue_file_test.go @@ -127,6 +127,67 @@ func TestOperationLockRecordsExactControlledSessionOwnership(t *testing.T) { } } +func TestOperationLockRecordsControlledSessionNetworkBeforeContainers(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 _, err := lock.AdmitLiveRunV1(LiveRunV1{ + ID: runID, Kind: LiveRunKindShellV1, Name: "controlled-session", + GenerationReference: generation, Exclusive: true, + }, false); err != nil { + t.Fatal(err) + } + complete := controlledSessionOwnershipFixtureV1(dir, runID, generation) + complete.NetworkName = "reploy-session-" + runID + complete.NetworkID = strings.Repeat("d", 64) + planned := complete + planned.NetworkID = "" + planned.Controller.ID = "" + planned.Workload.ID = "" + if _, err := lock.RecordControlledSessionOwnershipV1(planned); err != nil { + t.Fatal(err) + } + containerFirst := planned + containerFirst.Controller.ID = complete.Controller.ID + if _, err := lock.RecordControlledSessionOwnershipV1(containerFirst); err == nil || !strings.Contains(err.Error(), "before the network ID") { + t.Fatalf("container-first ownership error = %v", err) + } + networkPrepared := planned + networkPrepared.NetworkID = complete.NetworkID + recorded, err := lock.RecordControlledSessionOwnershipV1(networkPrepared) + if err != nil || recorded.NetworkID != complete.NetworkID || recorded.Controller.ID != "" { + t.Fatalf("network ownership = %#v, error=%v", recorded, err) + } + controllerPrepared := complete + controllerPrepared.Workload.ID = "" + recorded, err = lock.RecordControlledSessionOwnershipV1(controllerPrepared) + if err != nil || recorded.Controller.ID != complete.Controller.ID || recorded.Workload.ID != "" { + t.Fatalf("controller ownership = %#v, error=%v", recorded, err) + } + recorded, err = lock.RecordControlledSessionOwnershipV1(complete) + if err != nil { + t.Fatalf("complete network ownership = %#v, error=%v", recorded, err) + } + if recorded.NetworkID != complete.NetworkID || recorded.NetworkName != complete.NetworkName || recorded.Workload.ID != complete.Workload.ID { + t.Fatalf("recorded network ownership = %#v", recorded) + } + changedID := complete + changedID.NetworkID = strings.Repeat("e", 64) + if _, err := lock.RecordControlledSessionOwnershipV1(changedID); err == nil || !strings.Contains(err.Error(), "network ID changed") { + t.Fatalf("changed network ID error = %v", err) + } + changedName := complete + changedName.NetworkName = "other-network" + if _, err := lock.RecordControlledSessionOwnershipV1(changedName); err == nil || !strings.Contains(err.Error(), "immutable resource plan changed") { + t.Fatalf("changed network name error = %v", err) + } +} + func TestLiveRunQueueV1DecodesLegacyControlledSessionWithoutDockerEndpoint(t *testing.T) { dir := t.TempDir() bootSession, err := CurrentBootSessionIDV1() diff --git a/internal/dockerdeploy/controlled_session_incident_test.go b/internal/dockerdeploy/controlled_session_incident_test.go index 65203f36..79c75869 100644 --- a/internal/dockerdeploy/controlled_session_incident_test.go +++ b/internal/dockerdeploy/controlled_session_incident_test.go @@ -2,6 +2,7 @@ package dockerdeploy import ( "path/filepath" + "reflect" "strings" "testing" "time" @@ -32,7 +33,7 @@ func TestControlledSessionIncidentRetrievalSurfaceIsReadOnlyAndAcknowledgedExpli } receipts, err := ListControlledSessionIncidentReceiptsV1(t.Context(), dir) - if err != nil || len(receipts) != 1 || receipts[0] != receipt { + if err != nil || len(receipts) != 1 || !reflect.DeepEqual(receipts[0], receipt) { t.Fatalf("retrieved receipts = %#v, error=%v", receipts, err) } removed, err := AcknowledgeControlledSessionIncidentReceiptV1(t.Context(), dir, runID) diff --git a/internal/dockerdeploy/controlled_session_network.go b/internal/dockerdeploy/controlled_session_network.go index 3fd82844..9f6b67db 100644 --- a/internal/dockerdeploy/controlled_session_network.go +++ b/internal/dockerdeploy/controlled_session_network.go @@ -15,9 +15,11 @@ import ( "strings" "sync" "time" + + "github.com/omry/reploy/internal/deploy" ) -const controlledSessionNetworkRoleV1 = "network" +const controlledSessionNetworkRoleV1 = deploy.ControlledSessionNetworkRoleV1 type dockerSessionNetworkBackendV1 struct { bind func(context.Context, CommandSpec, time.Duration) (CommandSpec, commandRunner, error) @@ -407,7 +409,7 @@ func (network *DockerSessionNetworkV1) inspectV1(ctx context.Context) (dockerSes err := network.backend.run(command, RunOptions{Context: ctx, Stdout: &output, Stderr: &output}) if err != nil { message := trimmedCommandOutput(output.String()) - if isMissingDockerNetworkErrorV1(err) || strings.Contains(strings.ToLower(message), "no such network") { + if isMissingDockerNetworkResponseV1(err, message) { return dockerSessionNetworkInspectionV1{}, false, nil } if message != "" { @@ -584,11 +586,16 @@ func parseDockerNetworkIDV1(output string) (string, error) { } func isMissingDockerNetworkErrorV1(err error) bool { + return isMissingDockerNetworkResponseV1(err, "") +} + +func isMissingDockerNetworkResponseV1(err error, output string) bool { if err == nil { return false } - message := strings.ToLower(err.Error()) - return strings.Contains(message, "no such network") || strings.Contains(message, "network") && strings.Contains(message, "not found") + message := strings.ToLower(err.Error() + " " + output) + return strings.Contains(message, "no such network") || strings.Contains(message, "no such object") || + strings.Contains(message, "network") && strings.Contains(message, "not found") } func (network *DockerSessionNetworkV1) commandV1(args ...string) CommandSpec { diff --git a/internal/dockerdeploy/controlled_session_supervisor_test.go b/internal/dockerdeploy/controlled_session_supervisor_test.go index 8dbe8ea2..93c48d9a 100644 --- a/internal/dockerdeploy/controlled_session_supervisor_test.go +++ b/internal/dockerdeploy/controlled_session_supervisor_test.go @@ -532,7 +532,9 @@ func TestRunControlledSessionV1RejectsCleanupResourcesNotInDurableOwnership(t *t ownership := controlledSessionOwnershipFromPlanV1(plan, controlledSessionTestDockerEndpointV1, controllerID, workloadID) ownership.BootSession = "boot-session" manifest, manifestErr := deploy.ControlledSessionCleanupManifestFromOwnership(ownership) - manifest.Networks = []string{"unrelated-network"} + manifest.Networks = []deploy.ControlledSessionNetworkOwnershipV1{{ + Role: deploy.ControlledSessionNetworkRoleV1, ID: strings.Repeat("e", 64), Name: "unrelated-network", + }} return manifest, manifestErr }, now: time.Now, diff --git a/internal/dockerdeploy/controlled_session_watchdog.go b/internal/dockerdeploy/controlled_session_watchdog.go index 59d6b08a..0a1651ee 100644 --- a/internal/dockerdeploy/controlled_session_watchdog.go +++ b/internal/dockerdeploy/controlled_session_watchdog.go @@ -10,6 +10,7 @@ import ( "net" "net/http" "os" + "reflect" "strings" "syscall" "time" @@ -39,6 +40,8 @@ type controlledSessionWatchdogCleanupBackendV1 struct { bindDockerEndpoint func(string) error inspectContainer func(context.Context, string) (map[string]string, bool, error) removeContainer func(context.Context, string) error + inspectNetwork func(context.Context, string) (controlledSessionWatchdogNetworkInspectionV1, bool, error) + removeNetwork func(context.Context, string) error removeChannel func(string) error dockerUnavailable func(context.Context) (bool, error) waitRetry func(time.Duration) @@ -46,9 +49,17 @@ type controlledSessionWatchdogCleanupBackendV1 struct { writeIncident func(deploy.ControlledSessionIncidentReceiptV1) error } +type controlledSessionWatchdogNetworkInspectionV1 struct { + ID string + Name string + Labels map[string]string + Members map[string]string +} + type controlledSessionWatchdogCleanupReportV1 struct { controller deploy.ControlledSessionIncidentResourceStatusV1 workload deploy.ControlledSessionIncidentResourceStatusV1 + networks []deploy.ControlledSessionIncidentResourceStatusV1 channel deploy.ControlledSessionIncidentResourceStatusV1 } @@ -95,9 +106,12 @@ func runControlledSessionWatchdogV1( if err := backend.bindDockerEndpoint(manifest.DockerEndpoint); err != nil { return fmt.Errorf("bind watchdog Docker endpoint: %w", err) } - if len(manifest.Networks) != 0 || len(manifest.Volumes) != 0 { + if len(manifest.Volumes) != 0 { return fmt.Errorf("cleanup manifest names resources unsupported by this watchdog") } + if len(manifest.Networks) != 0 && (backend.inspectNetwork == nil || backend.removeNetwork == nil) { + return fmt.Errorf("watchdog network cleanup backend is incomplete") + } if err := requireControlledSessionWatchdogBootV1(manifest, backend); err != nil { return err } @@ -207,11 +221,18 @@ func cleanupControlledSessionFromWatchdogWithReportV1( report := controlledSessionWatchdogCleanupReportV1{ controller: deploy.ControlledSessionIncidentResourceCleanupFailedV1, workload: deploy.ControlledSessionIncidentResourceCleanupFailedV1, + networks: make([]deploy.ControlledSessionIncidentResourceStatusV1, len(manifest.Networks)), channel: deploy.ControlledSessionIncidentResourceCleanupFailedV1, } + for index := range report.networks { + report.networks[index] = deploy.ControlledSessionIncidentResourceCleanupFailedV1 + } if backend.inspectContainer == nil || backend.removeContainer == nil || backend.removeChannel == nil { return report, fmt.Errorf("watchdog cleanup backend is incomplete") } + if len(manifest.Networks) != 0 && (backend.inspectNetwork == nil || backend.removeNetwork == nil) { + return report, fmt.Errorf("watchdog network cleanup backend is incomplete") + } if err := deploy.ValidateControlledSessionCleanupManifest(manifest); err != nil { return report, err } @@ -226,6 +247,13 @@ func cleanupControlledSessionFromWatchdogWithReportV1( } else { report.controller = deploy.ControlledSessionIncidentResourceVerifiedAbsentV1 } + for index, network := range manifest.Networks { + if err := cleanupControlledSessionWatchdogNetworkV1(ctx, manifest.LiveRunID, network, backend); err != nil { + cleanupErr = errors.Join(cleanupErr, err) + } else { + report.networks[index] = deploy.ControlledSessionIncidentResourceVerifiedAbsentV1 + } + } if err := backend.removeChannel(manifest.ChannelDirectory); err != nil { cleanupErr = errors.Join(cleanupErr, err) } else { @@ -257,16 +285,67 @@ func controlledSessionWatchdogIncidentReceiptV1( cleanupStatus = deploy.ControlledSessionIncidentCleanupFailedV1 recoveryAction = deploy.ControlledSessionIncidentRecoveryNextOperationV1 } + networks := make([]deploy.ControlledSessionIncidentNetworkV1, len(manifest.Networks)) + for index, network := range manifest.Networks { + networks[index] = deploy.ControlledSessionIncidentNetworkV1{ + Role: network.Role, ID: network.ID, Name: network.Name, CleanupStatus: report.networks[index], + } + if report.networks[index] != deploy.ControlledSessionIncidentResourceVerifiedAbsentV1 { + cleanupStatus = deploy.ControlledSessionIncidentCleanupFailedV1 + recoveryAction = deploy.ControlledSessionIncidentRecoveryNextOperationV1 + } + } return deploy.ControlledSessionIncidentReceiptV1{ Schema: deploy.ControlledSessionIncidentReceiptSchemaV1, LiveRunID: manifest.LiveRunID, BootSession: manifest.BootSession, RecordedAt: recordedAt.UTC().Format(time.RFC3339Nano), Trigger: deploy.ControlledSessionIncidentParentLostV1, Controller: container(manifest.Controller, report.controller), Workload: container(manifest.Workload, report.workload), + Networks: networks, ChannelCleanupStatus: report.channel, CleanupStatus: cleanupStatus, RecoveryAction: recoveryAction, } } +func cleanupControlledSessionWatchdogNetworkV1( + ctx context.Context, + liveRunID string, + network deploy.ControlledSessionNetworkOwnershipV1, + backend controlledSessionWatchdogCleanupBackendV1, +) error { + inspection, found, err := backend.inspectNetwork(ctx, network.ID) + if err != nil { + return fmt.Errorf("inspect controlled-session network %q: %w", network.ID, err) + } + if !found { + return nil + } + if inspection.ID != network.ID { + return fmt.Errorf("refuse to remove controlled-session network %q because Docker returned full ID %q", network.ID, inspection.ID) + } + if inspection.Name != network.Name { + return fmt.Errorf("refuse to remove controlled-session network %q because network name does not match", network.ID) + } + expectedLabels := map[string]string{ + "io.reploy.session.live-run": liveRunID, + "io.reploy.session.role": deploy.ControlledSessionNetworkRoleV1, + } + if !reflect.DeepEqual(inspection.Labels, expectedLabels) { + return fmt.Errorf("refuse to remove controlled-session network %q because ownership labels do not match", network.ID) + } + if len(inspection.Members) != 0 { + return fmt.Errorf("refuse to remove controlled-session network %q because it still has members", network.ID) + } + removeErr := backend.removeNetwork(ctx, network.ID) + _, stillFound, inspectErr := backend.inspectNetwork(ctx, network.ID) + if inspectErr != nil { + return errors.Join(removeErr, fmt.Errorf("verify controlled-session network %q removal: %w", network.ID, inspectErr)) + } + if stillFound { + return errors.Join(removeErr, fmt.Errorf("controlled-session network %q still exists after removal", network.ID)) + } + return nil +} + func cleanupControlledSessionWatchdogContainerV1( ctx context.Context, liveRunID string, @@ -322,6 +401,12 @@ func productionControlledSessionWatchdogCleanupBackendV1(receipt *os.File) contr removeContainer: func(ctx context.Context, containerID string) error { return dockerRun(CommandSpec{Name: "docker", Args: []string{"container", "rm", "--force", containerID}}, RunOptions{Context: ctx}) }, + inspectNetwork: func(ctx context.Context, networkID string) (controlledSessionWatchdogNetworkInspectionV1, bool, error) { + return inspectControlledSessionWatchdogNetworkV1(ctx, networkID, dockerRun) + }, + removeNetwork: func(ctx context.Context, networkID string) error { + return dockerRun(CommandSpec{Name: "docker", Args: []string{"network", "rm", networkID}}, RunOptions{Context: ctx}) + }, removeChannel: removeControlledSessionChannelDirectoryV1, dockerUnavailable: func(ctx context.Context) (bool, error) { return controlledSessionWatchdogDockerUnavailableV1(ctx, dockerEndpoint, probeControlledSessionWatchdogDockerV1) @@ -334,6 +419,50 @@ func productionControlledSessionWatchdogCleanupBackendV1(receipt *os.File) contr } } +func inspectControlledSessionWatchdogNetworkV1( + ctx context.Context, + networkID string, + run commandRunner, +) (controlledSessionWatchdogNetworkInspectionV1, bool, error) { + var output bytes.Buffer + err := run(CommandSpec{Name: "docker", Args: []string{ + "network", "inspect", "--format", "{{json .Id}} {{json .Name}} {{json .Labels}} {{json .Containers}}", networkID, + }}, RunOptions{Context: ctx, Stdout: &output, Stderr: &output}) + if err != nil { + message := strings.TrimSpace(output.String()) + if isMissingDockerNetworkResponseV1(err, message) { + return controlledSessionWatchdogNetworkInspectionV1{}, false, nil + } + if message != "" { + return controlledSessionWatchdogNetworkInspectionV1{}, false, fmt.Errorf("%w: %s", err, message) + } + return controlledSessionWatchdogNetworkInspectionV1{}, false, err + } + decoder := json.NewDecoder(bytes.NewReader(output.Bytes())) + var inspection controlledSessionWatchdogNetworkInspectionV1 + var members map[string]dockerSessionNetworkContainerV1 + if err := decoder.Decode(&inspection.ID); err != nil { + return controlledSessionWatchdogNetworkInspectionV1{}, false, fmt.Errorf("decode inspected network ID: %w", err) + } + if err := decoder.Decode(&inspection.Name); err != nil { + return controlledSessionWatchdogNetworkInspectionV1{}, false, fmt.Errorf("decode inspected network name: %w", err) + } + if err := decoder.Decode(&inspection.Labels); err != nil { + return controlledSessionWatchdogNetworkInspectionV1{}, false, fmt.Errorf("decode inspected network labels: %w", err) + } + if err := decoder.Decode(&members); err != nil { + return controlledSessionWatchdogNetworkInspectionV1{}, false, fmt.Errorf("decode inspected network members: %w", err) + } + inspection.Members = make(map[string]string, len(members)) + for id, member := range members { + inspection.Members[id] = member.Name + } + if inspection.ID != networkID { + return controlledSessionWatchdogNetworkInspectionV1{}, false, fmt.Errorf("Docker inspected network %q as unexpected full ID %q", networkID, inspection.ID) + } + return inspection, true, nil +} + type controlledSessionWatchdogDockerProbeV1 func(context.Context, string) error func controlledSessionWatchdogDockerUnavailableV1( diff --git a/internal/dockerdeploy/controlled_session_watchdog_test.go b/internal/dockerdeploy/controlled_session_watchdog_test.go index 5fdc3e45..7b6ab094 100644 --- a/internal/dockerdeploy/controlled_session_watchdog_test.go +++ b/internal/dockerdeploy/controlled_session_watchdog_test.go @@ -104,6 +104,130 @@ func TestControlledSessionWatchdogParentLossRemovesOnlyManifestResources(t *test } } +func TestControlledSessionWatchdogRemovesExactOwnedNetworkAfterContainers(t *testing.T) { + manifest := controlledSessionWatchdogManifestFixtureV1(t) + manifest.Networks = []deploy.ControlledSessionNetworkOwnershipV1{{ + Role: deploy.ControlledSessionNetworkRoleV1, ID: strings.Repeat("e", 64), Name: "reploy-session-network", + }} + content, err := deploy.EncodeControlledSessionCleanupManifest(manifest) + if err != nil { + t.Fatal(err) + } + networkExists := true + var operations []string + var receipt deploy.ControlledSessionIncidentReceiptV1 + backend := controlledSessionWatchdogCleanupBackendV1{ + currentBootSession: func() (string, error) { return manifest.BootSession, nil }, + bindDockerEndpoint: func(string) error { return nil }, + inspectContainer: func(_ context.Context, id string) (map[string]string, bool, error) { + operations = append(operations, "container:"+id) + return nil, false, nil + }, + removeContainer: func(context.Context, string) error { return errors.New("absent containers must not be removed") }, + inspectNetwork: func(_ context.Context, id string) (controlledSessionWatchdogNetworkInspectionV1, bool, error) { + operations = append(operations, "network:"+id) + if !networkExists { + return controlledSessionWatchdogNetworkInspectionV1{}, false, nil + } + return controlledSessionWatchdogNetworkInspectionV1{ + ID: id, Name: manifest.Networks[0].Name, + Labels: map[string]string{ + "io.reploy.session.live-run": manifest.LiveRunID, + "io.reploy.session.role": deploy.ControlledSessionNetworkRoleV1, + }, + Members: map[string]string{}, + }, true, nil + }, + removeNetwork: func(_ context.Context, id string) error { + operations = append(operations, "remove-network:"+id) + networkExists = false + return nil + }, + removeChannel: func(string) error { operations = append(operations, "channel"); return nil }, + now: func() time.Time { return time.Date(2026, 8, 10, 3, 0, 0, 0, time.UTC) }, + writeIncident: func(value deploy.ControlledSessionIncidentReceiptV1) error { receipt = value; return nil }, + } + if err := runControlledSessionWatchdogV1(bytes.NewReader(content), strings.NewReader(""), io.Discard, backend); err != nil { + t.Fatal(err) + } + want := []string{ + "container:" + manifest.Workload.ID, + "container:" + manifest.Controller.ID, + "network:" + manifest.Networks[0].ID, + "remove-network:" + manifest.Networks[0].ID, + "network:" + manifest.Networks[0].ID, + "channel", + } + if !reflect.DeepEqual(operations, want) || networkExists { + t.Fatalf("network cleanup operations = %#v, exists=%t", operations, networkExists) + } + if len(receipt.Networks) != 1 || receipt.Networks[0].ID != manifest.Networks[0].ID || + receipt.Networks[0].CleanupStatus != deploy.ControlledSessionIncidentResourceVerifiedAbsentV1 || + receipt.CleanupStatus != deploy.ControlledSessionIncidentCleanupSucceededV1 { + t.Fatalf("network incident receipt = %#v", receipt) + } +} + +func TestControlledSessionWatchdogRefusesMismatchedOrOccupiedNetwork(t *testing.T) { + for _, test := range []struct { + name string + mutate func(*controlledSessionWatchdogNetworkInspectionV1) + want string + }{ + {name: "labels", mutate: func(value *controlledSessionWatchdogNetworkInspectionV1) { + value.Labels["io.reploy.session.live-run"] = "run-ffffffffffffffff" + }, want: "ownership labels"}, + {name: "member", mutate: func(value *controlledSessionWatchdogNetworkInspectionV1) { + value.Members[strings.Repeat("f", 64)] = "unrelated" + }, want: "still has members"}, + } { + t.Run(test.name, func(t *testing.T) { + manifest := controlledSessionWatchdogManifestFixtureV1(t) + manifest.Networks = []deploy.ControlledSessionNetworkOwnershipV1{{ + Role: deploy.ControlledSessionNetworkRoleV1, ID: strings.Repeat("e", 64), Name: "reploy-session-network", + }} + removed := false + backend := controlledSessionWatchdogCleanupBackendV1{ + inspectContainer: func(context.Context, string) (map[string]string, bool, error) { return nil, false, nil }, + removeContainer: func(context.Context, string) error { return nil }, + inspectNetwork: func(_ context.Context, id string) (controlledSessionWatchdogNetworkInspectionV1, bool, error) { + inspection := controlledSessionWatchdogNetworkInspectionV1{ + ID: id, Name: manifest.Networks[0].Name, + Labels: map[string]string{ + "io.reploy.session.live-run": manifest.LiveRunID, + "io.reploy.session.role": deploy.ControlledSessionNetworkRoleV1, + }, + Members: map[string]string{}, + } + test.mutate(&inspection) + return inspection, true, nil + }, + removeNetwork: func(context.Context, string) error { removed = true; return nil }, + removeChannel: func(string) error { return nil }, + } + err := cleanupControlledSessionFromWatchdogV1(t.Context(), manifest, backend) + if err == nil || !strings.Contains(err.Error(), test.want) || removed { + t.Fatalf("network cleanup error=%v, removed=%t", err, removed) + } + }) + } +} + +func TestControlledSessionWatchdogRejectsIncompleteNetworkCleanupBackend(t *testing.T) { + manifest := controlledSessionWatchdogManifestFixtureV1(t) + manifest.Networks = []deploy.ControlledSessionNetworkOwnershipV1{{ + Role: deploy.ControlledSessionNetworkRoleV1, ID: strings.Repeat("e", 64), Name: "reploy-session-network", + }} + err := cleanupControlledSessionFromWatchdogV1(t.Context(), manifest, controlledSessionWatchdogCleanupBackendV1{ + inspectContainer: func(context.Context, string) (map[string]string, bool, error) { return nil, false, nil }, + removeContainer: func(context.Context, string) error { return nil }, + removeChannel: func(string) error { return nil }, + }) + if err == nil || !strings.Contains(err.Error(), "network cleanup backend is incomplete") { + t.Fatalf("incomplete network backend error = %v", err) + } +} + func TestControlledSessionWatchdogRetriesParentLossCleanupWhileDockerIsUnavailable(t *testing.T) { manifest := controlledSessionWatchdogManifestFixtureV1(t) content, err := deploy.EncodeControlledSessionCleanupManifest(manifest) @@ -277,6 +401,9 @@ func TestControlledSessionWatchdogRefusesMismatchedOwnership(t *testing.T) { func TestControlledSessionWatchdogIncidentReceiptRecordsOnlyBoundedFailureStatus(t *testing.T) { manifest := controlledSessionWatchdogManifestFixtureV1(t) + manifest.Networks = []deploy.ControlledSessionNetworkOwnershipV1{{ + Role: deploy.ControlledSessionNetworkRoleV1, ID: strings.Repeat("e", 64), Name: "reploy-session-network", + }} content, err := deploy.EncodeControlledSessionCleanupManifest(manifest) if err != nil { t.Fatal(err) @@ -293,7 +420,11 @@ func TestControlledSessionWatchdogIncidentReceiptRecordsOnlyBoundedFailureStatus return nil, false, nil }, removeContainer: func(context.Context, string) error { return nil }, - removeChannel: func(string) error { return nil }, + inspectNetwork: func(context.Context, string) (controlledSessionWatchdogNetworkInspectionV1, bool, error) { + return controlledSessionWatchdogNetworkInspectionV1{}, false, errors.New(sensitive) + }, + removeNetwork: func(context.Context, string) error { return nil }, + removeChannel: func(string) error { return nil }, dockerUnavailable: func(context.Context) (bool, error) { return false, nil }, @@ -312,6 +443,7 @@ func TestControlledSessionWatchdogIncidentReceiptRecordsOnlyBoundedFailureStatus t.Fatal(encodeErr) } if bytes.Contains(encoded, []byte(sensitive)) || receipt.Workload.CleanupStatus != deploy.ControlledSessionIncidentResourceCleanupFailedV1 || + len(receipt.Networks) != 1 || receipt.Networks[0].CleanupStatus != deploy.ControlledSessionIncidentResourceCleanupFailedV1 || receipt.CleanupStatus != deploy.ControlledSessionIncidentCleanupFailedV1 || receipt.RecoveryAction != deploy.ControlledSessionIncidentRecoveryNextOperationV1 { t.Fatalf("unsafe or incomplete incident receipt = %s", encoded) @@ -347,6 +479,28 @@ func TestControlledSessionWatchdogClassifiesOnlyUnreachableDockerEndpointsAsUnav } } +func TestControlledSessionCleanupInspectorsRecognizeDockerNetworkNotFound(t *testing.T) { + inspectors := []struct { + name string + inspect func(context.Context, string, commandRunner) (controlledSessionWatchdogNetworkInspectionV1, bool, error) + }{ + {name: "watchdog", inspect: inspectControlledSessionWatchdogNetworkV1}, + {name: "recovery", inspect: inspectControlledSessionRecoveryNetworkV1}, + } + for _, inspector := range inspectors { + t.Run(inspector.name, func(t *testing.T) { + const message = "Error response from daemon: network eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee not found" + _, found, err := inspector.inspect(t.Context(), strings.Repeat("e", 64), func(_ CommandSpec, options RunOptions) error { + _, _ = io.WriteString(options.Stderr, message) + return errors.New("exit status 1") + }) + if err != nil || found { + t.Fatalf("network-not-found inspection found=%t, error=%v", found, err) + } + }) + } +} + func TestControlledSessionWatchdogRejectsPriorBootBeforeReady(t *testing.T) { manifest := controlledSessionWatchdogManifestFixtureV1(t) content, err := deploy.EncodeControlledSessionCleanupManifest(manifest) diff --git a/internal/dockerdeploy/live_run_recovery.go b/internal/dockerdeploy/live_run_recovery.go index 2302c60b..edf593da 100644 --- a/internal/dockerdeploy/live_run_recovery.go +++ b/internal/dockerdeploy/live_run_recovery.go @@ -8,6 +8,7 @@ import ( "fmt" "io" "path/filepath" + "reflect" "strings" "time" @@ -149,10 +150,107 @@ func cleanupControlledSessionRecoveryV1( for _, container := range []deploy.ControlledSessionContainerOwnershipV1{ownership.Workload, ownership.Controller} { cleanupErr = errors.Join(cleanupErr, cleanupControlledSessionRecoveryContainerV1(ctx, ownership.LiveRunID, container, pinnedRun)) } + if ownership.NetworkName != "" { + cleanupErr = errors.Join(cleanupErr, cleanupControlledSessionRecoveryNetworkV1(ctx, ownership, pinnedRun)) + } cleanupErr = errors.Join(cleanupErr, removeControlledSessionChannelDirectoryV1(ownership.ChannelDirectory)) return cleanupErr } +func cleanupControlledSessionRecoveryNetworkV1( + ctx context.Context, + ownership deploy.ControlledSessionOwnershipV1, + run commandRunner, +) error { + target := ownership.NetworkID + if target == "" { + target = ownership.NetworkName + } + inspection, found, err := inspectControlledSessionRecoveryNetworkV1(ctx, target, run) + if err != nil { + return fmt.Errorf("inspect recovered controlled-session network %q: %w", target, err) + } + if !found { + return nil + } + if ownership.NetworkID != "" && inspection.ID != ownership.NetworkID { + return fmt.Errorf("refuse to remove recovered controlled-session network %q because Docker returned full ID %q", ownership.NetworkID, inspection.ID) + } + if inspection.Name != ownership.NetworkName { + return fmt.Errorf("refuse to remove recovered controlled-session network %q because network name does not match", inspection.ID) + } + expectedLabels := map[string]string{ + "io.reploy.session.live-run": ownership.LiveRunID, + "io.reploy.session.role": deploy.ControlledSessionNetworkRoleV1, + } + if !reflect.DeepEqual(inspection.Labels, expectedLabels) { + return fmt.Errorf("refuse to remove recovered controlled-session network %q because ownership labels do not match", inspection.ID) + } + if len(inspection.Members) != 0 { + return fmt.Errorf("refuse to remove recovered controlled-session network %q because it still has members", inspection.ID) + } + removeErr := run(CommandSpec{Name: "docker", Args: []string{"network", "rm", inspection.ID}}, RunOptions{Context: ctx}) + if removeErr != nil && !isMissingDockerNetworkResponseV1(removeErr, "") { + return fmt.Errorf("remove recovered controlled-session network %q: %w", inspection.ID, removeErr) + } + _, stillFound, inspectErr := inspectControlledSessionRecoveryNetworkV1(ctx, inspection.ID, run) + if inspectErr != nil { + return fmt.Errorf("verify recovered controlled-session network %q removal: %w", inspection.ID, inspectErr) + } + if stillFound { + return fmt.Errorf("recovered controlled-session network %q still exists after removal", inspection.ID) + } + return nil +} + +func inspectControlledSessionRecoveryNetworkV1( + ctx context.Context, + target string, + run commandRunner, +) (controlledSessionWatchdogNetworkInspectionV1, bool, error) { + var output bytes.Buffer + err := run(CommandSpec{Name: "docker", Args: []string{ + "network", "inspect", "--format", "{{json .Id}} {{json .Name}} {{json .Labels}} {{json .Containers}}", target, + }}, RunOptions{Context: ctx, Stdout: &output, Stderr: &output}) + if err != nil { + message := strings.TrimSpace(output.String()) + if isMissingDockerNetworkResponseV1(err, message) { + return controlledSessionWatchdogNetworkInspectionV1{}, false, nil + } + if message != "" { + return controlledSessionWatchdogNetworkInspectionV1{}, false, fmt.Errorf("%w: %s", err, message) + } + return controlledSessionWatchdogNetworkInspectionV1{}, false, err + } + decoder := json.NewDecoder(bytes.NewReader(output.Bytes())) + var inspection controlledSessionWatchdogNetworkInspectionV1 + var members map[string]dockerSessionNetworkContainerV1 + if err := decoder.Decode(&inspection.ID); err != nil { + return controlledSessionWatchdogNetworkInspectionV1{}, false, fmt.Errorf("decode recovered controlled-session network ID: %w", err) + } + parsed, err := parseDockerNetworkIDV1(inspection.ID) + if err != nil { + return controlledSessionWatchdogNetworkInspectionV1{}, false, fmt.Errorf("decode recovered controlled-session network ID: %w", err) + } + if parsed != inspection.ID { + return controlledSessionWatchdogNetworkInspectionV1{}, false, fmt.Errorf("recovered controlled-session network ID is not canonical") + } + if err := decoder.Decode(&inspection.Name); err != nil { + return controlledSessionWatchdogNetworkInspectionV1{}, false, fmt.Errorf("decode recovered controlled-session network name: %w", err) + } + if err := decoder.Decode(&inspection.Labels); err != nil { + return controlledSessionWatchdogNetworkInspectionV1{}, false, fmt.Errorf("decode recovered controlled-session network labels: %w", err) + } + if err := decoder.Decode(&members); err != nil { + return controlledSessionWatchdogNetworkInspectionV1{}, false, fmt.Errorf("decode recovered controlled-session network members: %w", err) + } + inspection.Members = make(map[string]string, len(members)) + for id, member := range members { + inspection.Members[id] = member.Name + } + return inspection, true, nil +} + func cleanupControlledSessionRecoveryContainerV1( ctx context.Context, liveRunID string, diff --git a/internal/dockerdeploy/live_run_recovery_test.go b/internal/dockerdeploy/live_run_recovery_test.go index df1f974a..899341ce 100644 --- a/internal/dockerdeploy/live_run_recovery_test.go +++ b/internal/dockerdeploy/live_run_recovery_test.go @@ -307,7 +307,7 @@ func TestRecoverLiveRunQueueV1PreservesDurableCrashReceiptAfterResourceCleanup(t t.Fatalf("queue after recovery = %#v, found=%t, error=%v", queue, found, err) } receipts, err := operation.ReadControlledSessionIncidentReceiptsV1() - if err != nil || len(receipts) != 1 || receipts[0] != receipt { + if err != nil || len(receipts) != 1 || !reflect.DeepEqual(receipts[0], receipt) { t.Fatalf("receipt after recovery = %#v, error=%v", receipts, err) } removed, err := operation.AcknowledgeControlledSessionIncidentReceiptV1(recorded.LiveRunID) @@ -357,17 +357,107 @@ func TestRecoverLiveRunQueueV1RetainsControlledSessionAfterLabelMismatchAndRetri } } +func TestRecoverLiveRunQueueV1DiscoversAndRemovesOwnedNetworkByFrozenName(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + operation, err := deploy.AcquireOperationLock(t.Context(), plan.Workload.DeploymentDirectory) + if err != nil { + t.Fatal(err) + } + defer operation.Unlock() + run := liveRunAdmissionFixtureV1(plan.LiveRunID, false) + run.Kind = deploy.LiveRunKindShellV1 + run.GenerationReference = plan.Workload.GenerationReference + if _, err := operation.AdmitLiveRunV1(run, false); err != nil { + t.Fatal(err) + } + ownership := controlledSessionOwnershipFromPlanV1(plan, controlledSessionTestDockerEndpointV1, "", "") + ownership.NetworkName = "reploy-session-" + plan.LiveRunID + recorded, err := operation.RecordControlledSessionOwnershipV1(ownership) + if err != nil { + t.Fatal(err) + } + resources := newControlledSessionRecoveryContainersV1(recorded) + resources.byID = map[string]*controlledSessionRecoveryContainerFixtureV1{} + resources.byName = map[string]*controlledSessionRecoveryContainerFixtureV1{} + if _, err := recoverLiveRunQueueV1(t.Context(), operation, nil, resources.run); err != nil { + t.Fatal(err) + } + if resources.network != nil { + t.Fatalf("recovered network remains = %#v", resources.network) + } + wantNetworkInspects := []string{recorded.NetworkName, strings.Repeat("e", 64)} + if !reflect.DeepEqual(resources.networkInspects, wantNetworkInspects) { + t.Fatalf("network inspect targets = %#v, want %#v", resources.networkInspects, wantNetworkInspects) + } + if _, found, err := operation.ReadLiveRunQueueV1(); err != nil || found { + t.Fatalf("recovered network ownership remains: found=%t, error=%v", found, err) + } +} + +func TestRecoverLiveRunQueueV1RetainsLabelMismatchedNetworkAndRetries(t *testing.T) { + plan := controlledSessionControllerIntegrationPlanV1(t, "test-image", []string{"/controller"}) + operation, err := deploy.AcquireOperationLock(t.Context(), plan.Workload.DeploymentDirectory) + if err != nil { + t.Fatal(err) + } + defer operation.Unlock() + run := liveRunAdmissionFixtureV1(plan.LiveRunID, false) + run.Kind = deploy.LiveRunKindShellV1 + run.GenerationReference = plan.Workload.GenerationReference + if _, err := operation.AdmitLiveRunV1(run, false); err != nil { + t.Fatal(err) + } + ownership := controlledSessionOwnershipFromPlanV1( + plan, controlledSessionTestDockerEndpointV1, dockerControllerTestContainerIDV1, dockerWorkloadTestContainerIDV1, + ) + ownership.NetworkID = strings.Repeat("e", 64) + ownership.NetworkName = "reploy-session-" + plan.LiveRunID + recorded, err := operation.RecordControlledSessionOwnershipV1(ownership) + if err != nil { + t.Fatal(err) + } + 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 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()) + } + queue, found, err := operation.ReadLiveRunQueueV1() + if err != nil || !found || len(queue.ControlledSessions) != 1 { + t.Fatalf("retained network ownership = %#v, found=%t, error=%v", queue.ControlledSessions, found, err) + } + resources.network.labels["io.reploy.session.live-run"] = recorded.LiveRunID + if _, err := recoverLiveRunQueueV1(t.Context(), operation, nil, resources.run); err != nil { + t.Fatal(err) + } + if resources.network != nil { + t.Fatalf("retried network remains = %#v", resources.network) + } +} + type controlledSessionRecoveryContainerFixtureV1 struct { id string name string labels map[string]string } +type controlledSessionRecoveryNetworkFixtureV1 struct { + id string + name string + labels map[string]string + members map[string]dockerSessionNetworkContainerV1 +} + type controlledSessionRecoveryContainersV1 struct { - byID map[string]*controlledSessionRecoveryContainerFixtureV1 - byName map[string]*controlledSessionRecoveryContainerFixtureV1 - inspects []string - endpoint string + byID map[string]*controlledSessionRecoveryContainerFixtureV1 + byName map[string]*controlledSessionRecoveryContainerFixtureV1 + inspects []string + endpoint string + network *controlledSessionRecoveryNetworkFixtureV1 + networkInspects []string } func newControlledSessionRecoveryContainersV1(ownership deploy.ControlledSessionOwnershipV1) *controlledSessionRecoveryContainersV1 { @@ -396,6 +486,20 @@ func newControlledSessionRecoveryContainersV1(ownership deploy.ControlledSession containers.byID[container.id] = container containers.byName[container.name] = container } + if ownership.NetworkName != "" { + networkID := ownership.NetworkID + if networkID == "" { + networkID = strings.Repeat("e", 64) + } + containers.network = &controlledSessionRecoveryNetworkFixtureV1{ + id: networkID, name: ownership.NetworkName, + labels: map[string]string{ + "io.reploy.session.live-run": ownership.LiveRunID, + "io.reploy.session.role": deploy.ControlledSessionNetworkRoleV1, + }, + members: map[string]dockerSessionNetworkContainerV1{}, + } + } return containers } @@ -433,5 +537,31 @@ func (containers *controlledSessionRecoveryContainersV1) run(spec CommandSpec, o delete(containers.byName, container.name) return nil } + if len(spec.Args) >= 2 && spec.Args[0] == "network" && spec.Args[1] == "inspect" { + target := spec.Args[len(spec.Args)-1] + containers.networkInspects = append(containers.networkInspects, target) + network := containers.network + if network == nil || target != network.id && target != network.name { + return errors.New("No such network") + } + labels, err := json.Marshal(network.labels) + if err != nil { + return err + } + members, err := json.Marshal(network.members) + if err != nil { + return err + } + _, err = fmt.Fprintf(options.Stdout, "%q %q %s %s", network.id, network.name, labels, members) + return err + } + if len(spec.Args) >= 2 && spec.Args[0] == "network" && spec.Args[1] == "rm" { + id := spec.Args[len(spec.Args)-1] + if containers.network == nil || containers.network.id != id { + return errors.New("No such network") + } + containers.network = nil + return nil + } return fmt.Errorf("unexpected controlled-session recovery command: %#v", spec) }