fix(runner): Stream End fixes

pkg/runner_manager/manager.go

@@ -763,6 +763,11 @@ func (m *Manager) endRunnerJob(ctx context.Context, job model.StreamRunnerJob, d
 		DiscardVod: ptr.Take(discardVoD),
 	})
 	if err != nil {
+		// a job the runner doesn't know is already over, which is what we want anyway
+		if status.Code(err) == codes.NotFound {
+			m.logger.Info("runner does not know job, considering it ended", "runner", job.RunnerHostname, "job", job.JobID)
+			return nil
+		}
 		return fmt.Errorf("request stream end for job %s: %w", job.JobID, err)
 	}
 	return nil

runner/handlers.go

@@ -26,12 +26,18 @@ func (r *Runner) RequestStream(_ context.Context, req *protobuf.StreamRequest) (
 	a := []actions.Action{
 		actions.Stream,
 		actions.StreamEnd,
+	}
+	vod := []actions.Action{
 		actions.MkVOD,
 		actions.CheckVoD,
 		actions.MkThumb,
 	}
+	// runs instead of vod if the stream is ended with discardVod
+	discard := []actions.Action{
+		actions.DiscardRecording,
+	}
 
-	jID := r.RunAction(a, data, r.log.With("stream_id", req.GetStreamId(), "stream_version", req.GetVersion(), "input", req.GetInput()))
+	jID := r.RunAction(a, vod, discard, data, r.log.With("stream_id", req.GetStreamId(), "stream_version", req.GetVersion(), "input", req.GetInput()))
 	r.log.Info("job added", "ID", jID)
 
 	return &protobuf.StreamResponse{JobId: ptr.Take(jID)}, nil
@@ -46,7 +52,7 @@ func (r *Runner) RequestStreamEnd(_ context.Context, req *protobuf.StreamEndRequ
 	r.jobsMu.Unlock()
 	if ok {
 		cancel()
-		return nil, nil
+		return &protobuf.StreamEndResponse{}, nil
 	}
-	return nil, status.Errorf(codes.NotFound, "stream not found")
+	return nil, status.Errorf(codes.NotFound, "job %s not found", req.GetJobId())
 }

runner/pkg/actions/actions.go

@@ -13,5 +13,6 @@ import (
 // An action takes a context ctx, may cancel the action.
 // Actions should use log for logging and notify for sending messages like their progress to gocast.
 // d contains data passed to the action and is used to pass data to the next actions.
-// Any error, the action returns will be logged. If that error is an AbortingError, the subsequent actions will be skipped.
+// Any error, the action returns will be logged. If that error is an AbortingError, the action is not
+// retried. The actions after it still run, RunAction relies on that to end streams early.
 type Action func(ctx context.Context, log *slog.Logger, notify chan *protobuf.Notification, d map[string]any, metrics *metrics.Broker) error

runner/pkg/actions/discard.go

@@ -0,0 +1,22 @@
+package actions
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"time"
+
+	"github.com/tum-dev/gocast/runner/pkg/metrics"
+	"github.com/tum-dev/gocast/runner/protobuf"
+)
+
+// DiscardRecording marks the live recording for deletion instead of turning it into a VoD.
+// It runs in place of MkVOD/CheckVoD/MkThumb when a stream was ended with discardVod.
+func DiscardRecording(_ context.Context, log *slog.Logger, _ chan *protobuf.Notification, d map[string]any, _ *metrics.Broker) error {
+	recordingDir, ok := d["recordingDir"].(string)
+	if !ok {
+		return AbortingError(fmt.Errorf("no recordingDir in context"))
+	}
+	log.Info("marking recording for deletion", "dir", recordingDir)
+	return writeManagementFile(recordingDir, fmt.Sprintf(".del-%d", time.Now().Unix()))
+}

runner/runner.go

@@ -178,7 +178,9 @@ func (r *Runner) InitApiGrpc() {
 	}
 }
 
-func (r *Runner) RunAction(a []actions.Action, data map[string]any, logger *slog.Logger) string {
+// RunAction runs a in the background and returns the id of the created job.
+// The actions in a run even after the job was cancelled, afterwards either vod or discard runs.
+func (r *Runner) RunAction(a, vod, discard []actions.Action, data map[string]any, logger *slog.Logger) string {
 	// create new context to avoid cancellation on grpc request termination
 	c, cancel := context.WithCancel(context.Background())
 	job := uuid.New().String()
@@ -195,7 +197,8 @@ func (r *Runner) RunAction(a []actions.Action, data map[string]any, logger *slog
 			r.jobsMu.Unlock()
 			r.JobCount <- -1
 		}()
-		for _, action := range a {
+
+		run := func(action actions.Action) {
 			for {
 				log := logger.With("action", getFunctionName(action)).With("job", job)
 				log.Info("running action")
@@ -206,26 +209,35 @@ func (r *Runner) RunAction(a []actions.Action, data map[string]any, logger *slog
 					log.Error("action error", "error", err) // use action specific logger
 					if actions.IsAbortingError(err) {
 						log.Info("action can't continue")
-						break // escape retry loop on unrecoverable error
+						return // escape retry loop on unrecoverable error
 					}
 				} else {
-					break // escape retry loop on no error
+					return // escape retry loop on no error
 				}
 			}
-			// VoD creation (MkVOD, CheckVoD, MkThumb) is intentionally skipped once the
-			// recording is discarded, right after StreamEnd notifies gocast the stream has ended.
-			r.jobsMu.Lock()
-			shouldDiscard := r.discard[job]
-			r.jobsMu.Unlock()
-			if shouldDiscard && reflect.ValueOf(action).Pointer() == reflect.ValueOf(actions.StreamEnd).Pointer() {
-				logger.With("job", job).Info("discarding recording, skipping VoD creation")
-				break
-			}
+		}
+
+		for _, action := range a {
+			run(action)
+		}
+		next := vod
+		if r.discarded(job) {
+			logger.With("job", job).Info("discarding recording, skipping VoD creation")
+			next = discard
+		}
+		for _, action := range next {
+			run(action)
 		}
 	}()
 	return job
 }
 
+func (r *Runner) discarded(job string) bool {
+	r.jobsMu.Lock()
+	defer r.jobsMu.Unlock()
+	return r.discard[job]
+}
+
 func (r *Runner) handleNotifications(ctx context.Context) {
 	bounded := retry.NewFibonacci(1 * time.Second)
 	bounded = retry.WithJitter(500*time.Millisecond, bounded)