diff --git a/cmd/joj3/main.go b/cmd/joj3/main.go index a872ea7..4bf5d50 100644 --- a/cmd/joj3/main.go +++ b/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) diff --git a/cmd/joj3/stage.go b/cmd/joj3/stage.go index 12fc926..a68dc69 100644 --- a/cmd/joj3/stage.go +++ b/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) } diff --git a/cmd/joj3/stage_test.go b/cmd/joj3/stage_test.go new file mode 100644 index 0000000..e83cd4b --- /dev/null +++ b/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) + } +} diff --git a/internal/executor/dummy/executor.go b/internal/executor/dummy/executor.go index a31c348..e573d88 100644 --- a/internal/executor/dummy/executor.go +++ b/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 } diff --git a/internal/executor/local/executor.go b/internal/executor/local/executor.go index d91c02e..7695f8e 100644 --- a/internal/executor/local/executor.go +++ b/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 } diff --git a/internal/executor/sandbox/convert.go b/internal/executor/sandbox/convert.go index c907e17..8087dba 100644 --- a/internal/executor/sandbox/convert.go +++ b/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 { diff --git a/internal/executor/sandbox/convert_test.go b/internal/executor/sandbox/convert_test.go new file mode 100644 index 0000000..b5181ad --- /dev/null +++ b/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) + } +} diff --git a/internal/executor/sandbox/executor.go b/internal/executor/sandbox/executor.go index 987382f..3c94326 100644 --- a/internal/executor/sandbox/executor.go +++ b/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 } diff --git a/internal/executor/sandbox/grpc.go b/internal/executor/sandbox/grpc.go index 6db82b0..fa841b8 100644 --- a/internal/executor/sandbox/grpc.go +++ b/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) { diff --git a/internal/executor/sandbox/meta.go b/internal/executor/sandbox/meta.go index 0a72560..f327eec 100644 --- a/internal/executor/sandbox/meta.go +++ b/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, }) } diff --git a/internal/stage/executor.go b/internal/stage/executor.go index 8b54854..48a1007 100644 --- a/internal/stage/executor.go +++ b/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) { diff --git a/internal/stage/run.go b/internal/stage/run.go index 418154d..4d058fa 100644 --- a/internal/stage/run.go +++ b/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 } diff --git a/internal/stage/run_test.go b/internal/stage/run_test.go new file mode 100644 index 0000000..3d9103c --- /dev/null +++ b/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) + } +}