aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
Diffstat
-rw-r--r--cmd/mirum-worker/client.go4+3 −1
-rw-r--r--internal/executor/environment.go63+63 −0
-rw-r--r--internal/executor/executor.go101+27 −74
-rw-r--r--internal/executor/host/host.go190+190 −0
-rw-r--r--internal/executor/host/maxrss_other.go11+11 −0
-rw-r--r--internal/executor/host/maxrss_unix.go27+27 −0
-rw-r--r--internal/executor/runtime.go125+125 −0
7 files changed, 446 insertions, 75 deletions
diff --git a/cmd/mirum-worker/client.go b/cmd/mirum-worker/client.go
index 1525904..e23063e 100644
--- a/cmd/mirum-worker/client.go
+++ b/cmd/mirum-worker/client.go
@@ -19,6 +19,8 @@ import (
"dimidiumlabs/mirum/internal/protocol"
"dimidiumlabs/mirum/internal/protocol/wirepb"
"dimidiumlabs/mirum/internal/protocol/wirepb/wirepbconnect"
+
+ _ "dimidiumlabs/mirum/internal/executor/host" // registers the host Runtime backend
)
type client struct {
@@ -98,7 +100,7 @@ func (c *client) work(ctx context.Context) error {
task := resp.Msg
slog.Info("task received", "id", task.Id, "repo", task.RepoFullName)
- execErr := executor.Run(task.CloneUrl, task.Branch)
+ execErr := executor.Run(ctx, task.CloneUrl, task.Branch)
result := &wirepb.TaskResult{TaskId: task.Id, Success: execErr == nil}
if execErr != nil {
diff --git a/internal/executor/environment.go b/internal/executor/environment.go
new file mode 100644
--- /dev/null
+++ b/internal/executor/environment.go
@@ -0,0 +1,63 @@
+// Copyright (c) 2026 Nikolay Govorov
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package executor
+
+import (
+ "context"
+ "fmt"
+ "os"
+
+ "go.starlark.net/starlark"
+)
+
+// SOTaskCtx is the `ctx` value passed to the Starlark project() function. It
+// exposes the task's Runtime to the build script.
+type SOTaskCtx struct {
+ ctx context.Context
+ rt Runtime
+}
+
+var _ starlark.HasAttrs = (*SOTaskCtx)(nil)
+
+func (c *SOTaskCtx) Type() string { return "ctx" }
+func (c *SOTaskCtx) Hash() (uint32, error) { return 0, fmt.Errorf("unhashable: ctx") }
+func (c *SOTaskCtx) Truth() starlark.Bool { return true }
+func (c *SOTaskCtx) String() string { return "ctx" }
+
+func (c *SOTaskCtx) Freeze() {}
+
+func (c *SOTaskCtx) AttrNames() []string { return []string{"shell"} }
+func (c *SOTaskCtx) Attr(name string) (starlark.Value, error) {
+ switch name {
+ case "shell":
+ return starlark.NewBuiltin("ctx.shell", c.shell), nil
+ }
+ return nil, nil
+}
+
+func (c *SOTaskCtx) shell(
+ thread *starlark.Thread,
+ fn *starlark.Builtin,
+ args starlark.Tuple,
+ kwargs []starlark.Tuple,
+) (starlark.Value, error) {
+ var cmd string
+ if err := starlark.UnpackPositionalArgs(fn.Name(), args, kwargs, 1, &cmd); err != nil {
+ return nil, err
+ }
+
+ res, err := c.rt.Exec(c.ctx, Command{
+ Args: []string{"bash", "-c", cmd},
+ Stdout: os.Stdout,
+ Stderr: os.Stderr,
+ })
+ if err != nil {
+ return nil, err
+ }
+ if res.Code != 0 {
+ return nil, fmt.Errorf("shell: %q exited with code %d", cmd, res.Code)
+ }
+
+ return starlark.None, nil
+}
diff --git a/internal/executor/executor.go b/internal/executor/executor.go
index 7ce8770..d0e749f 100644
--- a/internal/executor/executor.go
+++ b/internal/executor/executor.go
@@ -1,15 +1,14 @@
// Copyright (c) 2026 Nikolay Govorov
// SPDX-License-Identifier: AGPL-3.0-or-later
+// Runtime executing the Starlark pipeline in one of the supported runtimes
package executor
import (
"bytes"
+ "context"
"fmt"
"log/slog"
- "os"
- "os/exec"
- "path/filepath"
"go.starlark.net/starlark"
"go.starlark.net/syntax"
@@ -17,85 +16,39 @@ import (
const entry = ".mirum/project.star"
-func RunCmd(dir, name string, args ...string) (string, error) {
- cmd := exec.Command(name, args...)
- cmd.Dir = dir
-
- var buf bytes.Buffer
- cmd.Stdout = &buf
- cmd.Stderr = &buf
-
- err := cmd.Run()
- return buf.String(), err
-}
-
-type SOTaskCtx struct {
- dir string
-}
-
-var _ starlark.HasAttrs = (*SOTaskCtx)(nil)
-
-func (c *SOTaskCtx) String() string { return "ctx" }
-func (c *SOTaskCtx) Type() string { return "ctx" }
-func (c *SOTaskCtx) Freeze() {}
-func (c *SOTaskCtx) Truth() starlark.Bool { return true }
-func (c *SOTaskCtx) Hash() (uint32, error) { return 0, fmt.Errorf("unhashable: ctx") }
-func (c *SOTaskCtx) AttrNames() []string { return []string{"shell"} }
-
-func (c *SOTaskCtx) Attr(name string) (starlark.Value, error) {
- switch name {
- case "shell":
- return starlark.NewBuiltin("ctx.shell", c.shell), nil
- }
- return nil, nil
-}
-
-func (c *SOTaskCtx) shell(
- thread *starlark.Thread,
- fn *starlark.Builtin,
- args starlark.Tuple,
- kwargs []starlark.Tuple,
-) (starlark.Value, error) {
- var cmd string
- if err := starlark.UnpackPositionalArgs(fn.Name(), args, kwargs, 1, &cmd); err != nil {
- return nil, err
- }
-
- proc := exec.Command("bash", "-c", cmd)
- proc.Dir = c.dir
- proc.Stdout = os.Stdout
- proc.Stderr = os.Stderr
- err := proc.Run()
+// Run selects a Runtime, clones the repository into it, runs the Starlark
+// build script, and discards the Runtime afterwards.
+func Run(ctx context.Context, cloneURL, branch string) error {
+ rt, err := NewRuntime()
if err != nil {
- return nil, err
- }
-
- return starlark.None, nil
-}
-
-// Run clones the repository into a temporary directory, runs the Starlark
-// build script, and cleans up afterwards.
-func Run(cloneURL, branch string) error {
- dir, err := os.MkdirTemp("", "mirum-*")
- if err != nil {
- return fmt.Errorf("create workdir: %w", err)
+ return fmt.Errorf("create runtime: %w", err)
}
defer func() {
- if err := os.RemoveAll(dir); err != nil {
- slog.Warn("executor: cleanup failed", "dir", dir, "err", err)
+ if err := rt.Close(); err != nil {
+ slog.Warn("executor: runtime cleanup failed", "err", err)
}
}()
- if out, err := RunCmd(dir, "git", "clone", "--depth=1", "--branch", branch, cloneURL, "."); err != nil {
- return fmt.Errorf("git clone: %s: %w", out, err)
+ var clone bytes.Buffer
+ res, err := rt.Exec(ctx, Command{
+ Args: []string{"git", "clone", "--depth=1", "--branch", branch, cloneURL, "."},
+ Stdout: &clone,
+ Stderr: &clone,
+ })
+ if err != nil {
+ return fmt.Errorf("git clone: %w", err)
+ }
+ if res.Code != 0 {
+ return fmt.Errorf("git clone exited with code %d: %s", res.Code, clone.String())
}
- return runStarlark(dir)
-}
+ var script bytes.Buffer
+ if err := rt.FileRecv(ctx, entry, &script); err != nil {
+ return fmt.Errorf("read %s: %w", entry, err)
+ }
-func runStarlark(dir string) error {
thread := &starlark.Thread{Name: "mirum"}
- globals, err := starlark.ExecFileOptions(&syntax.FileOptions{}, thread, filepath.Join(dir, entry), nil, nil)
+ globals, err := starlark.ExecFileOptions(&syntax.FileOptions{}, thread, entry, script.Bytes(), nil)
if err != nil {
return err
}
@@ -109,7 +62,7 @@ func runStarlark(dir string) error {
return fmt.Errorf("%s: project is not a function", entry)
}
- ctx := &SOTaskCtx{dir: dir}
- _, err = starlark.Call(thread, fn, starlark.Tuple{ctx}, nil)
+ tctx := &SOTaskCtx{ctx: ctx, rt: rt}
+ _, err = starlark.Call(thread, fn, starlark.Tuple{tctx}, nil)
return err
}
diff --git a/internal/executor/host/host.go b/internal/executor/host/host.go
new file mode 100644
--- /dev/null
+++ b/internal/executor/host/host.go
@@ -0,0 +1,190 @@
+// Copyright (c) 2026 Nikolay Govorov
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+// Package host implements an executor.Runtime that runs task commands
+// directly on the worker. It provides no isolation — it is the fast path
+// for local iteration (mirum task) and trusted workloads.
+package host
+
+import (
+ "context"
+ "fmt"
+ "io"
+ "os"
+ "os/exec"
+ "path/filepath"
+ "runtime"
+ "time"
+
+ "dimidiumlabs/mirum/internal/executor"
+)
+
+// Host is an executor.Runtime backed by a temporary directory on the worker.
+// Commands run as child processes of the worker, with no sandboxing.
+type Host struct {
+ root string
+}
+
+var _ executor.Runtime = (*Host)(nil)
+
+// New creates a Host runtime rooted at a fresh temporary directory.
+func New() (*Host, error) {
+ root, err := os.MkdirTemp("", "mirum-host-*")
+ if err != nil {
+ return nil, fmt.Errorf("create work dir: %w", err)
+ }
+ return &Host{root: root}, nil
+}
+
+func init() {
+ executor.RegisterRuntime(executor.RuntimeBackend{
+ Name: "host",
+ Priority: executor.PriorityHost,
+ New: func() (executor.Runtime, error) { return New() },
+ })
+}
+
+// Platform reports the worker's own OS and architecture.
+func (h *Host) Platform() executor.Platform {
+ return executor.Platform{OS: runtime.GOOS, Arch: runtime.GOARCH}
+}
+
+// resolve maps an env-relative slash path to an absolute path inside root,
+// rejecting paths that would escape it.
+func (h *Host) resolve(p string) (string, error) {
+ if p == "" {
+ return h.root, nil
+ }
+ local, err := filepath.Localize(p)
+ if err != nil {
+ return "", fmt.Errorf("invalid path %q: %w", p, err)
+ }
+ return filepath.Join(h.root, local), nil
+}
+
+// Exec runs cmd as a child process of the worker.
+func (h *Host) Exec(ctx context.Context, cmd executor.Command) (executor.Result, error) {
+ if len(cmd.Args) == 0 {
+ return executor.Result{}, fmt.Errorf("exec: empty args")
+ }
+
+ dir, err := h.resolve(cmd.Dir)
+ if err != nil {
+ return executor.Result{}, err
+ }
+
+ c := exec.CommandContext(ctx, cmd.Args[0], cmd.Args[1:]...)
+ c.Dir = dir
+ c.Stdin = cmd.Stdin
+ c.Stdout = cmd.Stdout
+ c.Stderr = cmd.Stderr
+ if len(cmd.Env) > 0 {
+ c.Env = os.Environ()
+ for k, v := range cmd.Env {
+ c.Env = append(c.Env, k+"="+v)
+ }
+ }
+
+ start := time.Now()
+ runErr := c.Run()
+ wall := time.Since(start)
+
+ if ctx.Err() != nil {
+ return executor.Result{}, ctx.Err()
+ }
+
+ ps := c.ProcessState
+ if ps == nil {
+ // The process never started, e.g. the executable was not found.
+ return executor.Result{}, fmt.Errorf("exec %s: %w", cmd.Args[0], runErr)
+ }
+
+ return executor.Result{
+ Code: ps.ExitCode(),
+ Usage: executor.Usage{
+ Wall: wall,
+ CPUUser: ps.UserTime(),
+ CPUSystem: ps.SystemTime(),
+ MaxRSS: maxRSS(ps),
+ },
+ }, nil
+}
+
+// FileSend writes a single file into the work directory, creating parent
+// directories as needed.
+func (h *Host) FileSend(ctx context.Context, path string, r io.Reader) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ dst, err := h.resolve(path)
+ if err != nil {
+ return err
+ }
+ if err := os.MkdirAll(filepath.Dir(dst), 0o755); err != nil {
+ return err
+ }
+ f, err := os.Create(dst)
+ if err != nil {
+ return err
+ }
+ _, copyErr := io.Copy(f, r)
+ closeErr := f.Close()
+ if copyErr != nil {
+ return copyErr
+ }
+ return closeErr
+}
+
+// FileRecv streams a single file out of the work directory.
+func (h *Host) FileRecv(ctx context.Context, path string, w io.Writer) error {
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ src, err := h.resolve(path)
+ if err != nil {
+ return err
+ }
+ f, err := os.Open(src)
+ if err != nil {
+ return err
+ }
+ defer func() { _ = f.Close() }()
+ _, err = io.Copy(w, f)
+ return err
+}
+
+// FileList returns the env-relative paths of every file under dir.
+func (h *Host) FileList(ctx context.Context, dir string) ([]string, error) {
+ base, err := h.resolve(dir)
+ if err != nil {
+ return nil, err
+ }
+
+ var out []string
+ err = filepath.WalkDir(base, func(p string, d os.DirEntry, err error) error {
+ if err != nil {
+ return err
+ }
+ if err := ctx.Err(); err != nil {
+ return err
+ }
+ if d.IsDir() {
+ return nil
+ }
+ rel, err := filepath.Rel(h.root, p)
+ if err != nil {
+ return err
+ }
+ out = append(out, filepath.ToSlash(rel))
+ return nil
+ })
+ if err != nil {
+ return nil, err
+ }
+ return out, nil
+}
+
+// Close removes the work directory.
+func (h *Host) Close() error {
+ return os.RemoveAll(h.root)
+}
diff --git a/internal/executor/host/maxrss_other.go b/internal/executor/host/maxrss_other.go
new file mode 100644
--- /dev/null
+++ b/internal/executor/host/maxrss_other.go
@@ -0,0 +1,11 @@
+// Copyright (c) 2026 Nikolay Govorov
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+//go:build !unix
+
+package host
+
+import "os"
+
+// maxRSS returns 0 on platforms without POSIX getrusage.
+func maxRSS(*os.ProcessState) int64 { return 0 }
diff --git a/internal/executor/host/maxrss_unix.go b/internal/executor/host/maxrss_unix.go
new file mode 100644
--- /dev/null
+++ b/internal/executor/host/maxrss_unix.go
@@ -0,0 +1,27 @@
+// Copyright (c) 2026 Nikolay Govorov
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+//go:build unix
+
+package host
+
+import (
+ "os"
+ "runtime"
+ "syscall"
+)
+
+// maxRSS extracts the peak resident set size of a finished process,
+// normalized to bytes. POSIX getrusage reports ru_maxrss in kilobytes on
+// Linux and the BSDs, but in bytes on macOS.
+func maxRSS(ps *os.ProcessState) int64 {
+ ru, ok := ps.SysUsage().(*syscall.Rusage)
+ if !ok {
+ return 0
+ }
+ rss := int64(ru.Maxrss)
+ if runtime.GOOS != "darwin" {
+ rss *= 1024
+ }
+ return rss
+}
diff --git a/internal/executor/runtime.go b/internal/executor/runtime.go
new file mode 100644
--- /dev/null
+++ b/internal/executor/runtime.go
@@ -0,0 +1,125 @@
+// Copyright (c) 2026 Nikolay Govorov
+// SPDX-License-Identifier: AGPL-3.0-or-later
+
+package executor
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+ "sort"
+ "time"
+)
+
+// Platform identifies the OS and architecture of a Runtime, as observed by
+// the task itself — the values behind tctx.os and tctx.arch.
+type Platform struct {
+ OS string // "linux", "darwin", "windows", ...
+ Arch string // "amd64", "arm64", "riscv64", ...
+}
+
+// Command is a single process to run inside a Runtime.
+type Command struct {
+ Args []string // argv; Args[0] is the executable
+ Dir string // working directory, relative to the env root
+ Env map[string]string // variables added to the environment's own
+ Stdin io.Reader // standard input; nil means none
+ Stdout io.Writer // streamed as produced; nil means discard
+ Stderr io.Writer // streamed as produced; nil means discard
+}
+
+// Usage reports the resources a Command consumed. It is collected from
+// POSIX rusage — os.ProcessState.SysUsage on the Host runtime, wait4 inside
+// the VM for the Qemu runtime — and feeds log output and billing metrics. A
+// backend leaves a field zero when it cannot measure it.
+type Usage struct {
+ Wall time.Duration // wall-clock time from start to exit
+ CPUUser time.Duration // CPU time spent in user mode
+ CPUSystem time.Duration // CPU time spent in kernel mode
+ MaxRSS int64 // peak resident set size, in bytes
+}
+
+// Result is the outcome of a Command that ran to completion.
+type Result struct {
+ Code int // process exit code; 0 means success
+ Usage Usage // resources the command consumed
+}
+
+// Runtime is an isolation backend: the environment in which one task's
+// commands run. The executor drives every Runtime identically and is
+// unaware of how isolation is achieved. A Runtime hosts one task and is
+// then discarded.
+type Runtime interface {
+ // Platform reports the environment's OS and architecture.
+ Platform() Platform
+
+ // Exec runs cmd to completion. The error covers failures to start or
+ // communicate with the process; a process that runs and exits non-zero
+ // is a successful Exec with a non-zero Result.Code.
+ Exec(ctx context.Context, cmd Command) (Result, error)
+
+ // List returns the env-relative paths of the files under dir, so the
+ // executor can resolve upload globs and walk a source tree.
+ FileList(ctx context.Context, dir string) ([]string, error)
+
+ // Send streams a single file into the environment at the env-relative
+ // path, creating parent directories as needed. Bytes flow straight from
+ // r — the worker never stages them on its own filesystem.
+ FileSend(ctx context.Context, path string, r io.Reader) error
+
+ // Recv streams a single file out of the environment to w. Like Send,
+ // nothing is staged on the worker's filesystem.
+ FileRecv(ctx context.Context, path string, w io.Writer) error
+
+ // Close discards the environment and releases its resources.
+ Close() error
+}
+
+// Selection priorities for RuntimeBackend.Priority; NewRuntime prefers higher.
+const (
+ PriorityHost = 0 // unisolated, runs on the worker itself
+ PriorityVM = 100 // hardware-isolated guest (QEMU)
+)
+
+// A RuntimeBackend describes one Runtime implementation. Backends register
+// themselves with RegisterRuntime, so the executor never imports the backend
+// packages directly.
+type RuntimeBackend struct {
+ Name string
+ Priority int
+
+ // New builds a fresh Runtime for one task. A nil Runtime with nil error
+ // means the backend does not apply on this worker, so NewRuntime falls
+ // through to the next.
+ New func() (Runtime, error)
+}
+
+// runtimeBackends is kept sorted by descending priority.
+var runtimeBackends []RuntimeBackend
+
+// RegisterRuntime adds a Runtime backend. Backends call it from an init
+// function.
+func RegisterRuntime(b RuntimeBackend) {
+ runtimeBackends = append(runtimeBackends, b)
+ sort.SliceStable(runtimeBackends, func(i, j int) bool {
+ return runtimeBackends[i].Priority > runtimeBackends[j].Priority
+ })
+}
+
+// NewRuntime builds a fresh Runtime for one task, choosing the highest-priority
+// backend that applies. The caller must Close the returned Runtime.
+func NewRuntime() (Runtime, error) {
+ for _, b := range runtimeBackends {
+ rt, err := b.New()
+ if err != nil {
+ return nil, fmt.Errorf("runtime %s: %w", b.Name, err)
+ }
+ if rt != nil {
+ slog.Info("executor: runtime selected", "backend", b.Name)
+ return rt, nil
+ }
+ }
+ return nil, errors.New("executor: no applicable runtime backend")
+}
generated by Gilti