From fe436af772f74fe054b6d2a72f41b7a5cfa7b064 Mon Sep 17 00:00:00 2001 From: Jon Langevin Date: Wed, 14 Oct 2026 22:23:07 -0400 Subject: [PATCH 1/2] fix(wfctl): preserve Docker resolver failure evidence --- cmd/wfctl/pipeline_docker_unix.go | 31 ++- cmd/wfctl/pipeline_record_diagnostic_unix.go | 56 +++++ .../pipeline_record_diagnostic_unix_test.go | 226 +++++++++++++++++- 3 files changed, 307 insertions(+), 6 deletions(-) diff --git a/cmd/wfctl/pipeline_docker_unix.go b/cmd/wfctl/pipeline_docker_unix.go index 59b74294..8b1571e6 100644 --- a/cmd/wfctl/pipeline_docker_unix.go +++ b/cmd/wfctl/pipeline_docker_unix.go @@ -61,7 +61,13 @@ type pipelineDockerProcessGroupKey struct{} var pipelineDockerTLSNames = [3]string{"ca.pem", "cert.pem", "key.pem"} -func resolvePipelineDockerClient(ctx context.Context) (*pipelineDockerClient, error) { +func resolvePipelineDockerClient(ctx context.Context) (_ *pipelineDockerClient, resolveErr error) { + phase := "docker_resolve_cli" + defer func() { + if resolveErr != nil { + pipelineRecordDiagnosticFromContext(ctx).fail(phase) + } + }() // Only discovery receives a snapshot of Docker's transport-selection env. // All daemon calls use a private empty config, explicit host/TLS flags, and // a minimal env, avoiding auth helpers, config proxies, and context changes. @@ -78,10 +84,12 @@ func resolvePipelineDockerClient(ctx context.Context) (*pipelineDockerClient, er if err != nil { return nil, fmt.Errorf("pipeline Docker CLI unavailable") } + phase = "docker_resolve_home" home := snapshot["HOME"] if home == "" { return nil, fmt.Errorf("pipeline Docker home unavailable") } + phase = "docker_resolve_config" config := snapshot["DOCKER_CONFIG"] if config == "" { config = filepath.Join(home, ".docker") @@ -97,14 +105,17 @@ func resolvePipelineDockerClient(ctx context.Context) (*pipelineDockerClient, er discoveryEnv = append(discoveryEnv, key+"="+value) } } - out, err := runPipelineDockerCLI(ctx, binary, discoveryEnv, pipelineDockerCallTimeout, "--config", config, "context", "inspect") + phase = "docker_resolve_context" + out, err := runPipelineDockerCLI(pipelineDockerResolutionContext(ctx, "resolve_context"), binary, discoveryEnv, pipelineDockerCallTimeout, "--config", config, "context", "inspect") if err != nil { return nil, fmt.Errorf("pipeline Docker context resolution failed") } + phase = "docker_resolve_context_metadata" var contexts []pipelineDockerContext if json.Unmarshal([]byte(out.Stdout), &contexts) != nil || len(contexts) != 1 { return nil, fmt.Errorf("pipeline Docker context metadata invalid") } + phase = "docker_resolve_endpoint" metadata := contexts[0] endpoint, ok := metadata.Endpoints["docker"] if !ok || metadata.Name == "" { @@ -113,6 +124,7 @@ func resolvePipelineDockerClient(ctx context.Context) (*pipelineDockerClient, er if err := validatePipelineDockerEndpoint(endpoint.Host); err != nil { return nil, err } + phase = "docker_resolve_tls" var material [3][]byte tlsEnabled := false verifyTLS := false @@ -153,6 +165,7 @@ func resolvePipelineDockerClient(ctx context.Context) (*pipelineDockerClient, er if (len(material[1]) == 0) != (len(material[2]) == 0) { return nil, fmt.Errorf("pipeline Docker TLS client material incomplete") } + phase = "docker_resolve_private_config" dir, err := os.MkdirTemp("", "wfctl-pipeline-docker-") if err != nil { return nil, fmt.Errorf("pipeline Docker private config unavailable") @@ -182,7 +195,8 @@ func resolvePipelineDockerClient(ctx context.Context) (*pipelineDockerClient, er c.args = append(c.args, flag+"="+path) } } - serverID, err := c.serverID(ctx) + phase = "docker_resolve_identity_metadata" + serverID, err := c.serverID(pipelineDockerResolutionContext(ctx, "resolve_identity")) if err != nil { return nil, err } @@ -426,13 +440,22 @@ func runPipelineDockerCLI(ctx context.Context, binary string, env []string, time if cmd.ProcessState != nil { result.ExitCode = cmd.ProcessState.ExitCode() } - if ctx.Err() != nil { + if contextErr := ctx.Err(); contextErr != nil { + outcome := "command" + if contextErr == context.DeadlineExceeded { + outcome = "deadline" + } else if contextErr == context.Canceled { + outcome = "cancelled" + } + observePipelineDockerResolutionFailure(ctx, outcome, result.ExitCode) return result, fmt.Errorf("pipeline Docker operation canceled or timed out") } if stdout.overflow || stderr.overflow { + observePipelineDockerResolutionFailure(ctx, "capture", result.ExitCode) return result, fmt.Errorf("pipeline Docker output exceeded capture limit") } if err != nil { + observePipelineDockerResolutionFailure(ctx, "command", result.ExitCode) return result, fmt.Errorf("pipeline Docker operation failed") } return result, nil diff --git a/cmd/wfctl/pipeline_record_diagnostic_unix.go b/cmd/wfctl/pipeline_record_diagnostic_unix.go index dee6ca42..2b023cee 100644 --- a/cmd/wfctl/pipeline_record_diagnostic_unix.go +++ b/cmd/wfctl/pipeline_record_diagnostic_unix.go @@ -43,6 +43,14 @@ var pipelineRecordDiagnosticPhases = map[string]bool{ "container_create": true, "container_id": true, "container_attach": true, "container_wait": true, "container_status": true, "container_command": true, "cleanup": true, "pipeline_result": true, "record_write": true, + "docker_resolve_cli": true, "docker_resolve_home": true, "docker_resolve_config": true, + "docker_resolve_context": true, "docker_resolve_context_metadata": true, + "docker_resolve_endpoint": true, "docker_resolve_tls": true, + "docker_resolve_private_config": true, "docker_resolve_identity_metadata": true, + "docker_resolve_context_cancelled": true, "docker_resolve_context_deadline": true, + "docker_resolve_context_capture": true, "docker_resolve_context_command": true, + "docker_resolve_identity_cancelled": true, "docker_resolve_identity_deadline": true, + "docker_resolve_identity_capture": true, "docker_resolve_identity_command": true, } type pipelineRecordDiagnostic struct { @@ -63,6 +71,54 @@ func pipelineRecordDiagnosticFromContext(ctx context.Context) *pipelineRecordDia return diagnostic } +// The private tag applies only to the two initial resolver calls. Subsequent +// server-ID checks and successful resolution never set operation/status fields. +type pipelineDockerResolutionKey struct{} + +func pipelineDockerResolutionContext(ctx context.Context, operation string) context.Context { + if pipelineRecordDiagnosticFromContext(ctx) == nil { + return ctx + } + switch operation { + case "resolve_context", "resolve_identity": + return context.WithValue(ctx, pipelineDockerResolutionKey{}, operation) + default: + return ctx + } +} + +func observePipelineDockerResolutionFailure(ctx context.Context, outcome string, exit int) { + diagnostic := pipelineRecordDiagnosticFromContext(ctx) + if diagnostic == nil { + return + } + operation, _ := ctx.Value(pipelineDockerResolutionKey{}).(string) + var stage string + switch operation { + case "resolve_context": + stage = "context" + case "resolve_identity": + stage = "identity" + default: + return + } + switch outcome { + case "cancelled", "deadline", "capture", "command": + default: + return + } + diagnostic.mu.Lock() + defer diagnostic.mu.Unlock() + if diagnostic.report.Phase == "" { + // Freeze the whole observation together; later cleanup or operations + // cannot mix their status with the first resolver failure. + diagnostic.report.Phase = "docker_resolve_" + stage + "_" + outcome + diagnostic.report.Operation = operation + diagnostic.report.DockerExit = pipelineRecordDiagnosticExit(exit) + diagnostic.report.ContainerExit = -1 + } +} + func (d *pipelineRecordDiagnostic) fail(phase string) { if d == nil { return diff --git a/cmd/wfctl/pipeline_record_diagnostic_unix_test.go b/cmd/wfctl/pipeline_record_diagnostic_unix_test.go index 3ef48a3f..02c044d7 100644 --- a/cmd/wfctl/pipeline_record_diagnostic_unix_test.go +++ b/cmd/wfctl/pipeline_record_diagnostic_unix_test.go @@ -87,10 +87,21 @@ func safePipelineRecordDiagnostic(stderr []byte) (pipelineRecordDiagnosticReport return pipelineRecordDiagnosticReport{}, "invalid" } switch report.Operation { - case "unknown", "create", "start", "wait": + case "unknown", "create", "start", "wait", "resolve_context", "resolve_identity": default: return pipelineRecordDiagnosticReport{}, "invalid" } + var resolutionOperation string + switch report.Phase { + case "docker_resolve_context_cancelled", "docker_resolve_context_deadline", "docker_resolve_context_capture", "docker_resolve_context_command": + resolutionOperation = "resolve_context" + case "docker_resolve_identity_cancelled", "docker_resolve_identity_deadline", "docker_resolve_identity_capture", "docker_resolve_identity_command": + resolutionOperation = "resolve_identity" + } + if (resolutionOperation != "" && report.Operation != resolutionOperation) || + ((report.Operation == "resolve_context" || report.Operation == "resolve_identity") && report.Operation != resolutionOperation) { + return pipelineRecordDiagnosticReport{}, "invalid" + } return report, "present" } @@ -410,7 +421,7 @@ func TestPipelineRecordDiagnosticsActualHostFailureAndSilence(t *testing.T) { if status != "missing" { t.Fatal("diagnostics emitted on success or without exact opt-in") } - } else if status != "present" || report.Phase != tc.phase || !report.ChildObserved || report.Cleanup != "complete" { + } else if status != "present" || report.Phase != tc.phase || !report.ChildObserved || report.Cleanup != "complete" || report.Operation != "unknown" || report.DockerExit != -1 { t.Fatalf("actual host failure lost phase/child/cleanup evidence: status=%s report=%+v", status, report) } if bytes.Contains(stderr, []byte("fixture-cli-stdout-canary")) { @@ -643,3 +654,214 @@ func TestPipelineRecordDiagnosticsContextClassification(t *testing.T) { } } } + +func TestPipelineRecordDiagnosticsResolverFailures(t *testing.T) { + for _, tc := range []struct { + name, fail, flood, metadata, server, phase, operation string + exit, calls int + }{ + {name: "context-command", fail: "context", phase: "docker_resolve_context_command", operation: "resolve_context", exit: 7, calls: 1}, + {name: "identity-command", fail: "info", phase: "docker_resolve_identity_command", operation: "resolve_identity", exit: 7, calls: 2}, + {name: "context-capture", flood: "context", phase: "docker_resolve_context_capture", operation: "resolve_context", exit: 0, calls: 1}, + {name: "identity-capture", flood: "info", phase: "docker_resolve_identity_capture", operation: "resolve_identity", exit: 0, calls: 2}, + {name: "context-metadata", metadata: "private-context-metadata-canary", phase: "docker_resolve_context_metadata", operation: "unknown", exit: -1, calls: 1}, + {name: "endpoint", metadata: `[{"Name":"default","Endpoints":{"docker":{"Host":"private-endpoint-canary"}}}]`, phase: "docker_resolve_endpoint", operation: "unknown", exit: -1, calls: 1}, + {name: "tls-metadata", metadata: `[{"Name":"private-context-canary","Endpoints":{"docker":{"Host":"tcp://localhost:2376"}},"TLSMaterial":{"docker":["private-tls-canary"]}}]`, phase: "docker_resolve_tls", operation: "unknown", exit: -1, calls: 1}, + {name: "identity-metadata", server: "private-server-canary / invalid", phase: "docker_resolve_identity_metadata", operation: "unknown", exit: -1, calls: 2}, + {name: "private-config", phase: "docker_resolve_private_config", operation: "unknown", exit: -1, calls: 1}, + } { + t.Run(tc.name, func(t *testing.T) { + directory, fixture := newPipelineDockerFixture(t) + fixture.Fail, fixture.Flood = tc.fail, tc.flood + if tc.metadata != "" { + fixture.Context = tc.metadata + } + if tc.server != "" { + fixture.ServerID = tc.server + } + if tc.name == "private-config" { + t.Setenv("TMPDIR", filepath.Join(directory, "nonexistent")) + } + writePipelineDockerFixture(t, directory, fixture) + diagnostic := newPipelineRecordDiagnostic() + ctx := context.WithValue(t.Context(), pipelineRecordDiagnosticKey{}, diagnostic) + client, err := resolvePipelineDockerClient(ctx) + if client != nil { + _ = client.Close() + } + if err == nil || client != nil { + t.Fatal("injected resolver failure was accepted") + } + report, output := diagnosticTestReport(t, diagnostic) + if report.Phase != tc.phase || report.Operation != tc.operation || report.DockerExit != tc.exit || report.ContainerExit != -1 || report.ChildObserved || report.Cleanup != "not_started" { + t.Fatal("resolver report lost actual failure stage/status origin") + } + if len(pipelineDockerCalls(t, directory)) != tc.calls { + t.Fatal("resolver diagnostics added or omitted an existing CLI call") + } + for _, secret := range []string{"TEST_SECRET_TOKEN", "raw daemon stderr", "canary", directory, "daemon-A", "localhost"} { + if bytes.Contains(output, []byte(secret)) { + t.Fatal("resolver report reflected private input or output") + } + } + if tc.name == "context-command" { + for _, invalid := range [][]byte{ + bytes.Replace(output, []byte(`"resolve_context"`), []byte(`"private-operation-canary"`), 1), + bytes.Replace(output, []byte(`"resolve_context"`), []byte(`"resolve_identity"`), 1), + bytes.Replace(output, []byte(`"resolve_context"`), []byte(`"unknown"`), 1), + bytes.Replace(output, []byte(`"docker_resolve_context_command"`), []byte(`"child_exit"`), 1), + bytes.Replace(output, []byte(`"container_exit":-1`), []byte(`"container_exit":0`), 1), + } { + if _, status := safePipelineRecordDiagnostic(invalid); status == "present" { + t.Fatal("safe consumer accepted an unknown or incoherent resolver status") + } + } + } + }) + } +} + +func TestPipelineRecordDiagnosticsResolverCancellationAndDeadline(t *testing.T) { + for _, stage := range []struct{ command, phase string }{{"context", "context"}, {"info", "identity"}} { + for _, outcome := range []string{"cancelled", "deadline"} { + t.Run(stage.phase+"-"+outcome, func(t *testing.T) { + directory, fixture := newPipelineDockerFixture(t) + fixture.Sleep = stage.command + writePipelineDockerFixture(t, directory, fixture) + diagnostic := newPipelineRecordDiagnostic() + outer, cancel := context.WithCancel(t.Context()) + defer cancel() + ctx := context.WithValue(outer, pipelineRecordDiagnosticKey{}, diagnostic) + stop := make(chan struct{}) + observed := make(chan bool, 1) + if outcome == "cancelled" { + // Cancel only after the real selected helper has logged its + // invocation, so identity cancellation cannot hit context discovery. + go func() { + deadline := time.NewTimer(10 * time.Second) + defer deadline.Stop() + tick := time.NewTicker(10 * time.Millisecond) + defer tick.Stop() + for { + select { + case <-stop: + observed <- false + return + case <-deadline.C: + cancel() + observed <- false + return + case <-tick.C: + data, _ := os.ReadFile(filepath.Join(directory, "calls.jsonl")) + for _, line := range bytes.Split(data, []byte{'\n'}) { + var call pipelineDockerInvocation + if json.Unmarshal(line, &call) == nil { + for _, arg := range call.Args { + if arg == stage.command { + cancel() + observed <- true + return + } + } + } + } + } + } + }() + } + client, err := resolvePipelineDockerClient(ctx) + close(stop) + if outcome == "cancelled" && !<-observed { + t.Fatal("cancellation control did not observe the selected real resolver call") + } + if client != nil { + _ = client.Close() + } + if err == nil || client != nil { + t.Fatal("resolver accepted the interrupted command") + } + var output bytes.Buffer + diagnostic.emit(outer, &output) + report, status := safePipelineRecordDiagnostic(output.Bytes()) + if status != "present" || report.Phase != "docker_resolve_"+stage.phase+"_"+outcome || report.Operation != "resolve_"+stage.phase || report.DockerExit != -1 || report.ContainerExit != -1 { + t.Fatal("interrupted resolver command lost its actual closed outcome") + } + if report.Cancelled != (outcome == "cancelled") || report.Deadline || (outcome == "deadline" && outer.Err() != nil) { + t.Fatal("inner command deadline was confused with outer context state") + } + }) + } + } +} + +func TestPipelineRecordDiagnosticsActualHostResolverFailure(t *testing.T) { + root := pipelineRecordHostTempDir(t) + binary := filepath.Join(root, "wfctl") + buildPipelineRecordHostBinary(t, binary, ".") + for _, enabled := range []string{"", "true", "1"} { + t.Run("opt-in-"+enabled, func(t *testing.T) { + directory, fixture := newPipelineDockerFixture(t) + fixture.Fail = "context" + writePipelineDockerFixture(t, directory, fixture) + t.Setenv("XDG_STATE_HOME", filepath.Join(pipelineRecordHostTempDir(t), "state")) + t.Setenv(pipelineRecordDiagnosticEnv, enabled) + stdout, stderr, err := runPipelineRecordHost(t, binary, "testdata/pipeline-record/builtin.yaml", "") + assertPipelineRecordHostError(t, stdout, stderr, err, "pipeline_failed") + report, status := safePipelineRecordDiagnostic(stderr) + if enabled == "1" { + if status != "present" || report.Phase != "docker_resolve_context_command" || report.Operation != "resolve_context" || report.DockerExit != 7 || report.ChildObserved || report.Cleanup != "not_started" { + t.Fatal("actual record parent did not consume the resolver observation") + } + } else if status != "missing" { + t.Fatal("resolver emitted diagnostics without literal opt-in") + } + for _, secret := range []string{"TEST_SECRET_TOKEN", "raw daemon stderr", directory} { + if bytes.Contains(stderr, []byte(secret)) { + t.Fatal("actual record parent exposed private resolver stderr") + } + } + }) + } +} + +func TestPipelineRecordDiagnosticsResolverFirstFailureAndConcurrency(t *testing.T) { + for _, first := range []bool{false, true} { + t.Run(map[bool]string{false: "concurrent-first", true: "preserve-first"}[first], func(t *testing.T) { + directory, fixture := newPipelineDockerFixture(t) + fixture.Fail, fixture.Flood = "context", "info" + writePipelineDockerFixture(t, directory, fixture) + diagnostic := newPipelineRecordDiagnostic() + ctx := context.WithValue(t.Context(), pipelineRecordDiagnosticKey{}, diagnostic) + if first { + if client, err := resolvePipelineDockerClient(ctx); err == nil || client != nil { + t.Fatal("real first resolver failure was not injected") + } + } + var workers sync.WaitGroup + outcomes := make(chan bool, 8) + for i := 0; i < 8; i++ { + workers.Go(func() { + operation, args, exit := "resolve_context", []string{"context", "inspect"}, 7 + if i%2 == 1 { + operation, args, exit = "resolve_identity", []string{"info", "--format", "{{json .ID}}"}, 0 + } + result, err := runPipelineDockerCLI(pipelineDockerResolutionContext(ctx, operation), filepath.Join(directory, "docker"), []string{"PATH=" + directory, "HOME=" + directory, "LANG=C", "LC_ALL=C"}, pipelineDockerCallTimeout, args...) + outcomes <- err != nil && result != nil && result.ExitCode == exit + }) + } + workers.Wait() + close(outcomes) + for ok := range outcomes { + if !ok { + t.Fatal("concurrent control missed a real command/capture failure") + } + } + report, _ := diagnosticTestReport(t, diagnostic) + contextFailure := report.Phase == "docker_resolve_context_command" && report.Operation == "resolve_context" && report.DockerExit == 7 + identityFailure := report.Phase == "docker_resolve_identity_capture" && report.Operation == "resolve_identity" && report.DockerExit == 0 + if (!contextFailure && !identityFailure) || (first && !contextFailure) || report.ContainerExit != -1 { + t.Fatal("concurrent observations mixed or rewrote the first failure tuple") + } + }) + } +} From 6e589784e4eb3ed6940b858417be79b8fbfacaec Mon Sep 17 00:00:00 2001 From: Jon Langevin Date: Wed, 14 Oct 2026 22:56:09 -0400 Subject: [PATCH 2/2] fix(wfctl): use tagged switch for resolver outcomes --- cmd/wfctl/pipeline_docker_unix.go | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/cmd/wfctl/pipeline_docker_unix.go b/cmd/wfctl/pipeline_docker_unix.go index 8b1571e6..55485c3f 100644 --- a/cmd/wfctl/pipeline_docker_unix.go +++ b/cmd/wfctl/pipeline_docker_unix.go @@ -442,9 +442,10 @@ func runPipelineDockerCLI(ctx context.Context, binary string, env []string, time } if contextErr := ctx.Err(); contextErr != nil { outcome := "command" - if contextErr == context.DeadlineExceeded { + switch contextErr { + case context.DeadlineExceeded: outcome = "deadline" - } else if contextErr == context.Canceled { + case context.Canceled: outcome = "cancelled" } observePipelineDockerResolutionFailure(ctx, outcome, result.ExitCode)