Report severed exec streams as execution_timeout, not success

.changes/unreleased/Bugfix-20260720-113539.yaml

@@ -0,0 +1,3 @@
+kind: Bugfix
+body: Report a job as execution_timeout instead of success when the pod exec stream is severed (pod evicted or exceeded its lifetime) before the command sequence completes. Previously such jobs were silently reported as successful with a missing outcome, so repo-check batches that timed out on large repos never retried.
+time: 2026-07-20T11:35:39.846412-04:00

src/pkg/k8s.go

@@ -1,12 +1,14 @@
 package pkg
 
 import (
+	"bytes"
 	"context"
 	"fmt"
 	"io"
 	"path"
 	"strings"
 	"sync"
+	"sync/atomic"
 	"time"
 
 	"github.com/rs/zerolog/log"
@@ -53,6 +55,9 @@ type JobConfig struct {
 	Stdin         io.Reader
 	Stdout        *SafeBuffer
 	Stderr        *SafeBuffer
+	// StdoutTee, when set, receives a copy of everything written to Stdout. It is
+	// used to watch for the job-completion marker without disturbing the log stream.
+	StdoutTee io.Writer
 }
 
 type JobRunner struct {
@@ -411,15 +416,38 @@ func (s *JobRunner) Run(ctx context.Context, job opslevel.RunnerJob, stdout, std
 	}
 
 	workingDirectory := path.Join(s.podConfig.WorkingDir, id)
-	commands := append([]string{fmt.Sprintf("mkdir -p %s", workingDirectory), fmt.Sprintf("cd %s", workingDirectory), "set -xv"}, job.Commands...)
-	runErr := s.Exec(ctx, stdout, stderr, pod, pod.Spec.Containers[0].Name, s.podConfig.Shell, "-e", "-c", strings.Join(commands, ";\n"))
+	// The marker is emitted from an EXIT trap so its presence in stdout proves the
+	// shell reached its own exit - covering normal completion and explicit `exit 0`
+	// (which some templates use), but not a pod that is killed/evicted mid-command
+	// (SIGKILL/torn-down streams don't run traps). On a `set -e` failure the trap
+	// still fires, but runErr is non-nil then so we report Failed before checking the
+	// marker. The marker embeds the unique pod identifier so job output can't spoof it.
+	marker := jobCompletionMarker(identifier)
+	commands := append([]string{
+		fmt.Sprintf("trap \"echo '%s'\" EXIT", marker),
+		fmt.Sprintf("mkdir -p %s", workingDirectory),
+		fmt.Sprintf("cd %s", workingDirectory),
+		"set -xv",
+	}, job.Commands...)
+	completed, runErr := s.Exec(ctx, stdout, stderr, marker, pod, pod.Spec.Containers[0].Name, s.podConfig.Shell, "-e", "-c", strings.Join(commands, ";\n"))
 	if runErr != nil {
 		return JobOutcome{
 			Message: fmt.Sprintf("pod execution failed REASON: %s %s", strings.TrimSuffix(stderr.String(), "\n"), runErr),
 			Outcome: opslevel.RunnerJobOutcomeEnumFailed,
 		}
 	}
 
+	// A nil runErr with no completion marker means the exec stream closed without the
+	// API server delivering an exit status - i.e. the pod was terminated (evicted or
+	// exceeded its lifetime) mid-command. Treat this as a timeout, not a success, so
+	// the (partial) outcome is not silently reported as complete and the job retries.
+	if !completed {
+		return JobOutcome{
+			Message: "job exec stream ended before the command sequence completed; the pod was likely terminated (evicted or exceeded its lifetime) before finishing",
+			Outcome: opslevel.RunnerJobOutcomeEnumExecutionTimeout,
+		}
+	}
+
 	return JobOutcome{
 		Message: "",
 		Outcome: opslevel.RunnerJobOutcomeEnumSuccess,
@@ -469,24 +497,94 @@ func (s *JobRunner) ExecWithConfig(ctx context.Context, config JobConfig) error
 	if err != nil {
 		return err
 	}
+	var stdout io.Writer = config.Stdout
+	if config.StdoutTee != nil {
+		stdout = io.MultiWriter(config.Stdout, config.StdoutTee)
+	}
 	return exec.StreamWithContext(ctx, remotecommand.StreamOptions{
 		Stdin:  config.Stdin,
-		Stdout: config.Stdout,
+		Stdout: stdout,
 		Stderr: config.Stderr,
 		Tty:    false,
 	})
 }
 
-func (s *JobRunner) Exec(ctx context.Context, stdout, stderr *SafeBuffer, pod *corev1.Pod, containerName string, cmd ...string) error {
-	return s.ExecWithConfig(ctx, JobConfig{
+// Exec runs cmd in the pod and reports whether the command sequence ran to
+// completion. Completion is proven by observing marker in stdout: it is emitted from
+// the shell's EXIT trap, so its absence on a clean (nil error) return means the exec
+// stream was severed before the shell exited. When marker is empty no detection is
+// performed and completed is reported true.
+func (s *JobRunner) Exec(ctx context.Context, stdout, stderr *SafeBuffer, marker string, pod *corev1.Pod, containerName string, cmd ...string) (completed bool, err error) {
+	var detector *markerWriter
+	config := JobConfig{
 		Command:       cmd,
 		Namespace:     pod.Namespace,
 		PodName:       pod.Name,
 		ContainerName: containerName,
 		Stdin:         nil,
 		Stdout:        stdout,
 		Stderr:        stderr,
-	})
+	}
+	if marker != "" {
+		detector = newMarkerWriter(marker)
+		config.StdoutTee = detector
+	}
+	err = s.ExecWithConfig(ctx, config)
+	if detector == nil {
+		return true, err
+	}
+	return detector.found(), err
+}
+
+// jobCompletionMarker returns the sentinel emitted from a job exec's EXIT trap. It
+// embeds the unique pod identifier so a job's own output cannot spoof it.
+func jobCompletionMarker(identifier string) string {
+	return fmt.Sprintf("[opslevel] job exec completed marker=%s", identifier)
+}
+
+// markerWriter is an io.Writer that watches the byte stream for a marker string,
+// tolerating the marker being split across Write calls. It retains at most
+// len(marker)-1 bytes between writes, so it is safe for arbitrarily large streams.
+type markerWriter struct {
+	marker []byte
+	carry  []byte
+	seen   atomic.Bool
+}
+
+func newMarkerWriter(marker string) *markerWriter {
+	return &markerWriter{marker: []byte(marker)}
+}
+
+func (w *markerWriter) Write(p []byte) (int, error) {
+	if !w.seen.Load() {
+		overlap := len(w.marker) - 1
+		// Check the seam spanning the previous chunk's tail and this chunk's head,
+		// then the chunk itself, avoiding a full-stream copy.
+		if len(w.carry) > 0 {
+			seam := append(append([]byte{}, w.carry...), p[:min(len(p), overlap)]...)
+			if bytes.Contains(seam, w.marker) {
+				w.seen.Store(true)
+			}
+		}
+		if !w.seen.Load() && bytes.Contains(p, w.marker) {
+			w.seen.Store(true)
+		}
+		if overlap > 0 {
+			tail := p
+			if len(tail) > overlap {
+				tail = tail[len(tail)-overlap:]
+			}
+			w.carry = append(w.carry, tail...)
+			if len(w.carry) > overlap {
+				w.carry = w.carry[len(w.carry)-overlap:]
+			}
+		}
+	}
+	return len(p), nil
+}
+
+func (w *markerWriter) found() bool {
+	return w.seen.Load()
 }
 
 func (s *JobRunner) CreateConfigMap(ctx context.Context, config *corev1.ConfigMap) (*corev1.ConfigMap, error) {

src/pkg/k8s_test.go

@@ -2,6 +2,7 @@ package pkg
 
 import (
 	"context"
+	"strings"
 	"testing"
 
 	"github.com/opslevel/opslevel-go/v2026"
@@ -314,5 +315,62 @@ func mountIsRW(mounts []corev1.VolumeMount, name string) bool {
 	return false
 }
 
+func TestJobCompletionMarker_EmbedsIdentifier(t *testing.T) {
+	// Arrange
+	identifier := "opslevel-job-abc-1700000000"
+	// Act
+	marker := jobCompletionMarker(identifier)
+	// Assert
+	autopilot.Assert(t, strings.Contains(marker, identifier), "marker must embed the unique pod identifier so job output cannot spoof it")
+}
+
+func TestMarkerWriter_DetectsMarkerInSingleWrite(t *testing.T) {
+	// Arrange
+	w := newMarkerWriter("MARKER")
+	// Act
+	_, err := w.Write([]byte("some logs\n[opslevel] MARKER done\nmore"))
+	// Assert
+	autopilot.Ok(t, err)
+	autopilot.Equals(t, true, w.found())
+}
+
+func TestMarkerWriter_DetectsMarkerSplitAcrossWrites(t *testing.T) {
+	// Arrange
+	w := newMarkerWriter("MARKER")
+	// Act: the marker straddles the boundary between two Write calls.
+	_, _ = w.Write([]byte("noise fooMAR"))
+	autopilot.Equals(t, false, w.found())
+	_, err := w.Write([]byte("KERbaz noise"))
+	// Assert
+	autopilot.Ok(t, err)
+	autopilot.Equals(t, true, w.found())
+}
+
+func TestMarkerWriter_AbsentMarker(t *testing.T) {
+	// Arrange
+	w := newMarkerWriter("MARKER")
+	// Act: a truncated stream that never emits the marker (severed exec).
+	_, _ = w.Write([]byte("processing check 83 of 100\n"))
+	_, err := w.Write([]byte("processing check 84 of 100\n"))
+	// Assert
+	autopilot.Ok(t, err)
+	autopilot.Equals(t, false, w.found())
+}
+
+func TestMarkerWriter_LargeStreamThenMarker(t *testing.T) {
+	// Arrange: ensure carry stays bounded and detection still works after lots of data.
+	w := newMarkerWriter("MARKER")
+	// Act
+	for i := 0; i < 1000; i++ {
+		_, _ = w.Write([]byte(strings.Repeat("x", 4096)))
+	}
+	autopilot.Equals(t, false, w.found())
+	autopilot.Assert(t, len(w.carry) < len(w.marker), "carry must never exceed len(marker)-1 bytes")
+	_, err := w.Write([]byte("tail MARKER tail"))
+	// Assert
+	autopilot.Ok(t, err)
+	autopilot.Equals(t, true, w.found())
+}
+
 // Suppress unused import warning for policyv1
 var _ = policyv1.PodDisruptionBudget{}