fix: harden stage execution failures

cmd/joj3/main.go

@@ -104,6 +104,7 @@ func run(conf *joj3Conf.Conf, conventionalCommit *joj3Conf.ConventionalCommit) e
 	)
 	if err != nil {
 		slog.Error("stage run", "error", err)
+		return err
 	}
 	if forceQuitStageName != "" {
 		slog.Info("stage force quit", "name", forceQuitStageName)

cmd/joj3/stage.go

@@ -1,7 +1,9 @@
 package main
 
 import (
+	"context"
 	"encoding/json"
+	"errors"
 	"fmt"
 	"log/slog"
 	"os"
@@ -124,6 +126,7 @@ func newErrorStageResults(err error) ([]stage.StageResult, string) {
 	}, "Internal Error"
 }
 
+//nolint:unparam // named err lets deferred cleanup errors reach the caller
 func runStages(
 	conf *conf.Conf,
 	groups []string,
@@ -153,36 +156,42 @@ func runStages(
 		stageResults, forceQuitStageName = newErrorStageResults(err)
 		return stageResults, forceQuitStageName, err
 	}
-	defer stage.Cleanup()
+	ctx := context.Background()
+	defer func() {
+		err = errors.Join(err, stage.Cleanup(ctx))
+	}()
 	// ignore force quit in preStages & postStages
 	slog.Info("run preStages")
-	_, _, err = stage.Run(preStages)
-	if err != nil {
-		slog.Error("run preStages", "error", err)
+	_, _, preErr := stage.Run(ctx, preStages)
+	if preErr != nil {
+		slog.Error("run preStages", "error", preErr)
 	}
 	slog.Info("run stages")
-	stageResults, forceQuitStageName, err = stage.Run(stages)
-	if err != nil {
-		slog.Error("run stages", "error", err)
-		stageResults, forceQuitStageName = newErrorStageResults(err)
+	stageResults, forceQuitStageName, mainErr := stage.Run(ctx, stages)
+	if mainErr != nil {
+		slog.Error("run stages", "error", mainErr)
+		stageResults, forceQuitStageName = newErrorStageResults(mainErr)
 	}
 	onStagesComplete(stageResults, forceQuitStageName)
 	slog.Info("output result start", "path", conf.OutputPath)
 	slog.Debug("output result start",
 		"path", conf.OutputPath, "results", stageResults)
-	content, err := json.Marshal(stageResults)
-	if err != nil {
-		slog.Error("marshal stageResults", "error", err)
+	content, marshalErr := json.Marshal(stageResults)
+	if marshalErr != nil {
+		slog.Error("marshal stageResults", "error", marshalErr)
 	}
-	err = os.WriteFile(conf.OutputPath,
-		append(content, []byte("\n")...), 0o600)
-	if err != nil {
-		slog.Error("write stageResults", "error", err)
+	var outputErr error
+	if marshalErr == nil {
+		outputErr = os.WriteFile(conf.OutputPath,
+			append(content, '\n'), 0o600)
+		if outputErr != nil {
+			slog.Error("write stageResults", "error", outputErr)
+		}
 	}
 	slog.Info("run postStages")
-	_, _, err = stage.Run(postStages)
-	if err != nil {
-		slog.Error("run postStages", "error", err)
+	_, _, postErr := stage.Run(ctx, postStages)
+	if postErr != nil {
+		slog.Error("run postStages", "error", postErr)
 	}
-	return stageResults, forceQuitStageName, err
+	return stageResults, forceQuitStageName, errors.Join(preErr, mainErr, marshalErr, outputErr, postErr)
 }

cmd/joj3/stage_test.go

@@ -0,0 +1,27 @@
+package main
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/joint-online-judge/JOJ3/cmd/joj3/conf"
+	"github.com/joint-online-judge/JOJ3/internal/stage"
+)
+
+func TestRunStagesPreservesMainAndOutputErrors(t *testing.T) {
+	brokenStage := conf.ConfStage{Name: "broken"}
+	brokenStage.Executor.Name = "missing-executor"
+	c := &conf.Conf{
+		SandboxExecServer: "localhost:5051",
+		OutputPath:        t.TempDir(),
+		Stages:            []conf.ConfStage{brokenStage},
+	}
+	_, _, err := runStages(c, nil, func([]stage.StageResult, string) {})
+	if err == nil {
+		t.Fatal("runStages() unexpectedly succeeded")
+	}
+	if !strings.Contains(err.Error(), "executor not found") ||
+		!strings.Contains(err.Error(), "is a directory") {
+		t.Fatalf("runStages() error = %v, want main and output errors", err)
+	}
+}

internal/executor/dummy/executor.go

@@ -1,8 +1,12 @@
 package dummy
 
-import "github.com/joint-online-judge/JOJ3/internal/stage"
+import (
+	"context"
 
-func (e *Dummy) Run(cmds []stage.Cmd) ([]stage.ExecutorResult, error) {
+	"github.com/joint-online-judge/JOJ3/internal/stage"
+)
+
+func (e *Dummy) Run(_ context.Context, cmds []stage.Cmd) ([]stage.ExecutorResult, error) {
 	res := make([]stage.ExecutorResult, 0, len(cmds))
 	for range cmds {
 		res = append(res, stage.ExecutorResult{
@@ -19,6 +23,6 @@ func (e *Dummy) Run(cmds []stage.Cmd) ([]stage.ExecutorResult, error) {
 	return res, nil
 }
 
-func (e *Dummy) Cleanup() error {
+func (e *Dummy) Cleanup(_ context.Context) error {
 	return nil
 }

internal/executor/local/executor.go

@@ -2,6 +2,7 @@ package local
 
 import (
 	"bytes"
+	"context"
 	"fmt"
 	"io"
 	"math"
@@ -105,10 +106,16 @@ func (e *Local) generateResult(
 	return result
 }
 
-func (e *Local) Run(cmds []stage.Cmd) ([]stage.ExecutorResult, error) {
+func (e *Local) Run(ctx context.Context, cmds []stage.Cmd) ([]stage.ExecutorResult, error) {
 	var results []stage.ExecutorResult
 
 	for _, cmd := range cmds {
+		if len(cmd.Args) == 0 {
+			return nil, fmt.Errorf("command args must not be empty")
+		}
+		if err := ctx.Err(); err != nil {
+			return nil, err
+		}
 		execCmd := exec.Command(cmd.Args[0], cmd.Args[1:]...) // #nosec G204
 		execCmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
 		if cmd.CPULimit > 0 && cmd.ClockLimit <= 0 {
@@ -167,6 +174,10 @@ func (e *Local) Run(cmds []stage.Cmd) ([]stage.ExecutorResult, error) {
 				false,
 			)
 			results = append(results, result)
+		case <-ctx.Done():
+			_ = syscall.Kill(-execCmd.Process.Pid, syscall.SIGKILL)
+			<-done
+			return nil, ctx.Err()
 		case <-time.After(duration):
 			_ = syscall.Kill(-execCmd.Process.Pid, syscall.SIGKILL)
 			err := <-done
@@ -216,6 +227,6 @@ func handleCopyOut(result *stage.ExecutorResult, cmd stage.Cmd) error {
 	return nil
 }
 
-func (e *Local) Cleanup() error {
+func (e *Local) Cleanup(_ context.Context) error {
 	return nil
 }

internal/executor/sandbox/convert.go

@@ -1,7 +1,7 @@
 package sandbox
 
 import (
-	"log/slog"
+	"fmt"
 	"os"
 	"path/filepath"
 	"strings"
@@ -12,14 +12,22 @@ import (
 )
 
 // copied from https://github.com/criyle/go-judge/blob/master/cmd/go-judge-shell/grpc.go
-func convertPBCmd(cmd []stage.Cmd) []*pb.Request_CmdType {
+func convertPBCmd(cmd []stage.Cmd) ([]*pb.Request_CmdType, error) {
 	ret := make([]*pb.Request_CmdType, 0, len(cmd))
-	for _, c := range cmd {
+	for index, c := range cmd {
+		files, err := convertPBFiles([]*stage.CmdFile{c.Stdin, c.Stdout, c.Stderr})
+		if err != nil {
+			return nil, fmt.Errorf("command %d standard file: %w", index, err)
+		}
+		copyIn, err := convertPBCopyIn(c.CopyIn, c.CopyInDir)
+		if err != nil {
+			return nil, fmt.Errorf("command %d copy-in: %w", index, err)
+		}
 		req := &pb.Request_CmdType{}
 		req.SetArgs(c.Args)
 		req.SetEnv(c.Env)
 		req.SetTty(c.TTY)
-		req.SetFiles(convertPBFiles([]*stage.CmdFile{c.Stdin, c.Stdout, c.Stderr}))
+		req.SetFiles(files)
 		req.SetCpuTimeLimit(c.CPULimit)
 		req.SetClockTimeLimit(c.ClockLimit)
 		req.SetMemoryLimit(c.MemoryLimit)
@@ -29,7 +37,7 @@ func convertPBCmd(cmd []stage.Cmd) []*pb.Request_CmdType {
 		req.SetCpuSetLimit(c.CPUSetLimit)
 		req.SetDataSegmentLimit(c.DataSegmentLimit)
 		req.SetAddressSpaceLimit(c.AddressSpaceLimit)
-		req.SetCopyIn(convertPBCopyIn(c.CopyIn, c.CopyInDir))
+		req.SetCopyIn(copyIn)
 		req.SetCopyOut(convertPBCopyOut(c.CopyOut))
 		req.SetCopyOutCached(convertPBCopyOut(c.CopyOutCached))
 		req.SetCopyOutMax(c.CopyOutMax)
@@ -37,25 +45,25 @@ func convertPBCmd(cmd []stage.Cmd) []*pb.Request_CmdType {
 		req.SetSymlinks(convertSymlink(c.CopyIn))
 		ret = append(ret, req)
 	}
-	return ret
+	return ret, nil
 }
 
 func convertPBCopyIn(
 	copyIn map[string]stage.CmdFile, copyInDir string,
-) map[string]*pb.Request_File {
+) (map[string]*pb.Request_File, error) {
 	if copyInDir != "" {
-		_ = filepath.Walk(copyInDir,
+		err := filepath.Walk(copyInDir,
 			func(path string, info os.FileInfo, err error) error {
 				if err != nil {
-					return nil
+					return err
 				}
 				absPath, err := filepath.Abs(path)
 				if err != nil {
-					return nil
+					return err
 				}
 				relPath, err := filepath.Rel(copyInDir, path)
 				if err != nil {
-					return nil
+					return err
 				}
 				_, exists := copyIn[relPath]
 				if !info.IsDir() && !exists {
@@ -63,15 +71,22 @@ func convertPBCopyIn(
 				}
 				return nil
 			})
+		if err != nil {
+			return nil, fmt.Errorf("walk %q: %w", copyInDir, err)
+		}
 	}
 	rt := make(map[string]*pb.Request_File, len(copyIn))
 	for k, i := range copyIn {
 		if i.Symlink != nil {
 			continue
 		}
-		rt[k] = convertPBFile(i)
+		file, err := convertPBFile(i)
+		if err != nil {
+			return nil, fmt.Errorf("file %q: %w", k, err)
+		}
+		rt[k] = file
 	}
-	return rt
+	return rt, nil
 }
 
 func convertPBCopyOut(copyOut []string) []*pb.Request_CmdCopyOutFile {
@@ -101,65 +116,67 @@ func convertSymlink(copyIn map[string]stage.CmdFile) map[string]string {
 	return ret
 }
 
-func convertPBFiles(files []*stage.CmdFile) []*pb.Request_File {
+func convertPBFiles(files []*stage.CmdFile) ([]*pb.Request_File, error) {
 	var ret []*pb.Request_File
 	for _, f := range files {
 		if f == nil {
 			ret = append(ret, nil)
 		} else {
-			ret = append(ret, convertPBFile(*f))
+			file, err := convertPBFile(*f)
+			if err != nil {
+				return nil, err
+			}
+			ret = append(ret, file)
 		}
 	}
-	return ret
+	return ret, nil
 }
 
-func convertPBFile(i stage.CmdFile) *pb.Request_File {
+func convertPBFile(i stage.CmdFile) (*pb.Request_File, error) {
 	req := &pb.Request_File{}
 	switch {
 	case i.Src != nil:
 		if !filepath.IsAbs(*i.Src) {
 			absPath, err := filepath.Abs(*i.Src)
 			if err != nil {
-				slog.Error("convert pb file get abs path", "path", *i.Src, "error", err)
-				absPath = "/"
+				return nil, fmt.Errorf("resolve source path %q: %w", *i.Src, err)
 			}
 			i.Src = &absPath
 		}
 		s, err := os.ReadFile(*i.Src)
 		if err != nil {
-			s = []byte{}
-			slog.Error("convert pb file read file", "path", *i.Src, "error", err)
+			return nil, fmt.Errorf("read source file %q: %w", *i.Src, err)
 		}
 		m := &pb.Request_MemoryFile{}
 		m.SetContent(s)
 		req.SetMemory(m)
-		return req
+		return req, nil
 	case i.Content != nil:
 		s := strToBytes(*i.Content)
 		m := &pb.Request_MemoryFile{}
 		m.SetContent(s)
 		req.SetMemory(m)
-		return req
+		return req, nil
 	case i.FileID != nil:
 		c := &pb.Request_CachedFile{}
 		c.SetFileID(*i.FileID)
 		req.SetCached(c)
-		return req
+		return req, nil
 	case i.Name != nil && i.Max != nil:
 		p := &pb.Request_PipeCollector{}
 		p.SetName(*i.Name)
 		p.SetMax(*i.Max)
 		p.SetPipe(i.Pipe)
 		req.SetPipe(p)
-		return req
+		return req, nil
 	case i.StreamIn:
 		req.SetStreamIn(&emptypb.Empty{})
-		return req
+		return req, nil
 	case i.StreamOut:
 		req.SetStreamOut(&emptypb.Empty{})
-		return req
+		return req, nil
 	}
-	return nil
+	return nil, nil
 }
 
 func convertPBResult(res []*pb.Response_Result) []stage.ExecutorResult {

internal/executor/sandbox/convert_test.go

@@ -0,0 +1,21 @@
+package sandbox
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/joint-online-judge/JOJ3/internal/stage"
+)
+
+func TestConvertPBCmdReturnsSourceReadError(t *testing.T) {
+	missing := t.TempDir() + "/missing"
+	_, err := convertPBCmd([]stage.Cmd{{
+		Args: []string{"true"},
+		CopyIn: map[string]stage.CmdFile{
+			"input": {Src: &missing},
+		},
+	}})
+	if err == nil || !strings.Contains(err.Error(), "read source file") {
+		t.Fatalf("convertPBCmd() error = %v, want source read error", err)
+	}
+}

internal/executor/sandbox/executor.go

@@ -2,6 +2,7 @@ package sandbox
 
 import (
 	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 	"maps"
@@ -11,11 +12,11 @@ import (
 	"google.golang.org/protobuf/proto"
 )
 
-func (e *Sandbox) Run(cmds []stage.Cmd) ([]stage.ExecutorResult, error) {
+func (e *Sandbox) Run(ctx context.Context, cmds []stage.Cmd) ([]stage.ExecutorResult, error) {
 	var err error
 	if e.execClient == nil {
 		slog.Debug("create exec client", "server", e.execServer)
-		e.execClient, err = createExecClient(e.execServer, e.token)
+		e.execClient, e.conn, err = createExecClient(e.execServer, e.token)
 		if err != nil {
 			return nil, err
 		}
@@ -32,14 +33,19 @@ func (e *Sandbox) Run(cmds []stage.Cmd) ([]stage.ExecutorResult, error) {
 			}
 		}
 	}
-	pbCmds := convertPBCmd(cmds)
+	pbCmds, err := convertPBCmd(cmds)
+	if err != nil {
+		return nil, err
+	}
 	for i, pbCmd := range pbCmds {
 		slog.Debug("sandbox execute", "i", i, "pbCmd size", proto.Size(pbCmd))
 	}
 	pbReq := &pb.Request{}
 	pbReq.SetCmd(pbCmds)
 	slog.Debug("sandbox execute", "pbReq size", proto.Size(pbReq))
-	pbRet, err := e.execClient.Exec(context.TODO(), pbReq)
+	callCtx, cancel := context.WithTimeout(ctx, e.timeout)
+	defer cancel()
+	pbRet, err := e.execClient.Exec(callCtx, pbReq)
 	if err != nil {
 		return nil, err
 	}
@@ -53,15 +59,24 @@ func (e *Sandbox) Run(cmds []stage.Cmd) ([]stage.ExecutorResult, error) {
 	return results, nil
 }
 
-func (e *Sandbox) Cleanup() error {
+func (e *Sandbox) Cleanup(ctx context.Context) error {
+	var cleanupErr error
 	for k, fileID := range e.cachedMap {
 		req := &pb.FileID{}
 		req.SetFileID(fileID)
-		_, err := e.execClient.FileDelete(context.TODO(), req)
+		callCtx, cancel := context.WithTimeout(ctx, e.timeout)
+		_, err := e.execClient.FileDelete(callCtx, req)
+		cancel()
 		if err != nil {
 			slog.Error("sandbox cleanup", "error", err)
+			cleanupErr = errors.Join(cleanupErr, err)
 		}
 		delete(e.cachedMap, k)
 	}
-	return nil
+	if e.conn != nil {
+		cleanupErr = errors.Join(cleanupErr, e.conn.Close())
+		e.conn = nil
+		e.execClient = nil
+	}
+	return cleanupErr
 }

internal/executor/sandbox/grpc.go

@@ -12,13 +12,13 @@ import (
 )
 
 // copied from https://github.com/criyle/go-judger-demo/blob/master/apigateway/main.go
-func createExecClient(execServer, token string) (pb.ExecutorClient, error) {
+func createExecClient(execServer, token string) (pb.ExecutorClient, *grpc.ClientConn, error) {
 	conn, err := createGRPCConnection(execServer, token)
 	if err != nil {
 		slog.Error("gRPC connection", "error", err)
-		return nil, err
+		return nil, nil, err
 	}
-	return pb.NewExecutorClient(conn), nil
+	return pb.NewExecutorClient(conn), conn, nil
 }
 
 func createGRPCConnection(addr, token string) (*grpc.ClientConn, error) {

internal/executor/sandbox/meta.go

@@ -5,8 +5,11 @@
 package sandbox
 
 import (
+	"time"
+
 	"github.com/criyle/go-judge/pb"
 	"github.com/joint-online-judge/JOJ3/internal/stage"
+	"google.golang.org/grpc"
 )
 
 var name = "sandbox"
@@ -15,6 +18,8 @@ type Sandbox struct {
 	execServer, token string
 	cachedMap         map[string]string
 	execClient        pb.ExecutorClient
+	conn              *grpc.ClientConn
+	timeout           time.Duration
 }
 
 func init() {
@@ -22,6 +27,7 @@ func init() {
 		execServer: "localhost:5051",
 		token:      "",
 		cachedMap:  make(map[string]string),
+		timeout:    30 * time.Second,
 	})
 }
 
@@ -31,5 +37,6 @@ func InitWithConf(execServer, token string) {
 		execServer: execServer,
 		token:      token,
 		cachedMap:  make(map[string]string),
+		timeout:    30 * time.Second,
 	})
 }

internal/stage/executor.go

@@ -1,6 +1,7 @@
 package stage
 
 import (
+	"context"
 	"encoding/json"
 	"fmt"
 	"strconv"
@@ -9,8 +10,8 @@ import (
 var executorMap = map[string]Executor{}
 
 type Executor interface {
-	Run([]Cmd) ([]ExecutorResult, error)
-	Cleanup() error
+	Run(context.Context, []Cmd) ([]ExecutorResult, error)
+	Cleanup(context.Context) error
 }
 
 func RegisterExecutor(name string, executor Executor) {

internal/stage/run.go

@@ -4,11 +4,13 @@
 package stage
 
 import (
+	"context"
+	"errors"
 	"fmt"
 	"log/slog"
 )
 
-func Run(stages []Stage) (
+func Run(ctx context.Context, stages []Stage) (
 	stageResults []StageResult, forceQuitStageName string, err error,
 ) {
 	var executorResults []ExecutorResult
@@ -59,9 +61,10 @@ func Run(stages []Stage) (
 					"name", stage.Executor.Name,
 				)
 				err = fmt.Errorf("executor not found: %s", stage.Executor.Name)
+				forceQuitStageName = stage.Name
 				return
 			}
-			executorResults, err = executor.Run(stage.Executor.Cmds)
+			executorResults, err = executor.Run(ctx, stage.Executor.Cmds)
 			if err != nil {
 				slog.Error(
 					"executor run error",
@@ -69,6 +72,7 @@ func Run(stages []Stage) (
 					"name", stage.Executor.Name,
 					"error", err,
 				)
+				forceQuitStageName = stage.Name
 				return
 			}
 			for i, executorResult := range executorResults {
@@ -115,6 +119,7 @@ func Run(stages []Stage) (
 						"name", stageParser.Name,
 					)
 					err = fmt.Errorf("parser not found: %s", stageParser.Name)
+					forceQuitStageName = stage.Name
 					return
 				}
 				var parserForceQuit bool
@@ -130,6 +135,14 @@ func Run(stages []Stage) (
 					forceQuitStageName = stage.Name
 					break
 				}
+				if len(tmpParserResults) != len(executorResults) {
+					err = fmt.Errorf(
+						"parser %q returned %d results for %d executor results",
+						stageParser.Name, len(tmpParserResults), len(executorResults),
+					)
+					forceQuitStageName = stage.Name
+					break
+				}
 				for i, parserResult := range tmpParserResults {
 					parserScoresMap[stageParser.Name][i] += parserResult.Score
 				}
@@ -185,12 +198,15 @@ func Run(stages []Stage) (
 	return stageResults, forceQuitStageName, err
 }
 
-func Cleanup() {
+func Cleanup(ctx context.Context) error {
 	slog.Info("stage cleanup start")
+	var cleanupErr error
 	for name, executor := range executorMap {
-		err := executor.Cleanup()
+		err := executor.Cleanup(ctx)
 		if err != nil {
 			slog.Error("executor cleanup error", "name", name, "error", err)
+			cleanupErr = errors.Join(cleanupErr, fmt.Errorf("executor %q cleanup: %w", name, err))
 		}
 	}
+	return cleanupErr
 }

internal/stage/run_test.go

@@ -0,0 +1,40 @@
+package stage
+
+import (
+	"context"
+	"strings"
+	"testing"
+)
+
+type contractTestExecutor struct{}
+
+func (contractTestExecutor) Run(context.Context, []Cmd) ([]ExecutorResult, error) {
+	return []ExecutorResult{{Status: StatusAccepted}}, nil
+}
+
+func (contractTestExecutor) Cleanup(context.Context) error { return nil }
+
+type contractTestParser struct{}
+
+func (contractTestParser) Run([]ExecutorResult, any) ([]ParserResult, bool, error) {
+	return nil, false, nil
+}
+
+func TestRunRejectsParserResultCountMismatch(t *testing.T) {
+	const executorName = "contract-test-executor"
+	const parserName = "contract-test-parser"
+	RegisterExecutor(executorName, contractTestExecutor{})
+	RegisterParser(parserName, contractTestParser{})
+
+	_, forceQuit, err := Run(context.Background(), []Stage{{
+		Name:     "test",
+		Executor: StageExecutor{Name: executorName, Cmds: []Cmd{{}}},
+		Parsers:  []StageParser{{Name: parserName}},
+	}})
+	if err == nil || !strings.Contains(err.Error(), "returned 0 results for 1") {
+		t.Fatalf("Run() error = %v, want result count error", err)
+	}
+	if forceQuit != "test" {
+		t.Fatalf("Run() force quit = %q, want test", forceQuit)
+	}
+}