openvibers/openvibe.link Commits

T14 step 1 — S4: accept and run the code runtime class in OpenVibe.Node (#23)

OP OpenVibers committed 4 Oct 2026 · 87cf94630d · 11 files, +228 −44

local checks ✓ 1/1 · CI ✓ 9/9 · review SHIP by DeepSeek harness · 1 fix round

https://github.com/OpenVibers/OpenVibe.Node/pull/23

Changes

ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex 6aa327c..3d43a58 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -181,13 +181,14 @@ Commands are never queued while offline and never replayed after a reconnect.
### Jobs
The job frames of OpenVibe.Contracts `platform.job-frame@1` (plan T14 Run) ride the same link and envelope. The Node
--runs class `function` only, and only when the owner turned the worker on in the local config, the OS is Linux and a
--boot-time probe could start a process in its own user, network and PID namespaces ([worker.md](worker.md)).
++runs classes `function` and `code` only: each only when the owner declared an entry of that class, turned the worker on
++in the local config, the OS is Linux and a boot-time probe could start a process in its own user, network and PID
++namespaces ([worker.md](worker.md)); both run in the same sandbox.
Otherwise it advertises no class and refuses every well-formed job with `class not available`, and nothing executes.
| direction | type | fields |
|------------------|----------------|-------------------------------------------------------------------------------------------|
--| server → device | `job` | `job` {`id` (`job_<ULID>`: the idempotency and ack key), `class`, `artifact` {`name`, `version`} (required for `function`, one exact version), `args` {}, `ttl_ms`, `limits` {`wall_ms`, `cpu_ms`, `mem_bytes`}, `net` (`deny`, the default)} |
++| server → device | `job` | `job` {`id` (`job_<ULID>`: the idempotency and ack key), `class`, `artifact` {`name`, `version`} (required for `function` and `code`, one exact version), `args` {}, `ttl_ms`, `limits` {`wall_ms`, `cpu_ms`, `mem_bytes`}, `net` (`deny`, the default)} |
| server → device | `job_cancel` | `id` |
| server → device | `job_exit_ack` | `id` |
| device → server | `job_started` | `id`, `started_ms` |
@@@ -203,7 +204,7 @@ checked in this order:
| `missing or malformed job id` | `bad_value` | `id` is not `job_` + 26 ULID characters (the nack echoes it) |
| `malformed job` | `bad_frame` | the `job` object does not decode (a field of the wrong type) |
| `unknown class` | `unsupported` | `class` is not one of `function`, `code`, `browser`, `linux`, `desktop`, `gpu` |
--| `function job needs an artifact with an exact version` | `bad_value` | a `function` job without `artifact`, or a name or version outside the contract's pattern (a range) |
++| `function or code job needs an artifact with an exact version` | `bad_value` | a `function` or `code` job without `artifact`, or a name or version outside the contract's pattern (a range) |
| `net policy not supported` | `unsupported` | `net` is present and not `deny` |
| `missing or invalid args, ttl_ms or limits` | `bad_value` | `args` is not an object, or `ttl_ms` or a limit is missing or below 1 |
| `class not available` | `unsupported` | the class is valid but this Node does not run it (every class but `function`; every class while the worker is off) |
@@@ -250,9 +251,9 @@ worker caps in `status.capabilities.worker`:
`{"capabilities":["worker:function"],"runtime_classes":["function"],"max_jobs":1,"max_ttl_ms":600000,"max_wall_ms":300000,"max_cpu_ms":300000,"max_mem_bytes":536870912,"max_output_bytes":1048576}`.
`capabilities` holds one `worker:<class>` name per entry in `runtime_classes` (the `platform.resource-offer@1`
capabilities namespace OpenVibe.Run routes on; the names are reserved in OpenVibe.Contracts and no schema change
--was needed), so only classes this Node actually runs are advertised: `worker:function` today, `worker:code` once the
--`code` class ships. The limits are `worker.caps` with defaults applied; the key is absent while the worker is off or
--declares no `functions` (a probe that passes with none advertises nothing, so Fabric places no job this Node would
++was needed), so only classes this Node actually runs are advertised: `worker:function` when a `function` entry is declared,
++`worker:code` when a `code` entry is (`"runtime_classes":["function","code"]`, in that order, with both). The limits are `worker.caps` with defaults applied; the key is absent while the worker is off or
++declares no `functions` entry (a probe that passes with none advertises nothing, so Fabric places no job this Node would
refuse), and `worker` is reserved as a plugin name.
### Node descriptor → Fabric offers
ddiff --git a/docs/worker.md b/docs/worker.md
iindex 84ef7a9..da1d787 100644
---- a/docs/worker.md
++++ b/docs/worker.md
@@@ -1,8 +1,8 @@
--# The job worker (`function` jobs)
++# The job worker (`function` and `code` jobs)
--The Node can run **function jobs** (OpenVibe.Contracts `platform.job@1`, class `function`) sent over the control link
--([protocol.md](protocol.md#jobs)). A job is untrusted input: it names a function, and the Node runs it only if the
--owner declared that function, at that exact version, in the local config. Nothing is downloaded and no path comes from
++The Node can run **function and code jobs** (OpenVibe.Contracts `platform.job@1`, class `function` or `code`) sent over
++the control link ([protocol.md](protocol.md#jobs)). A job is untrusted input: it names an artifact, and the Node runs it
++only if the owner declared that artifact, at that exact version and for that class, in the local config. Nothing is downloaded and no path comes from
the server. The worker is **off by default**; off, the Node refuses every job `class not available`, as before.
## Config
@@@ -14,7 +14,8 @@ the server. The worker is **off by default**; off, the Node refuses every job `c
"enabled": false,
"functions": [
{"name": "thumbnail", "version": "1.2.0", "command": ["/opt/fn/thumbnail/bin/run"], "env": {"MODE": "fast"},
-- "artifact_dir": "/opt/fn/thumbnail"}
++ "artifact_dir": "/opt/fn/thumbnail"},
++ {"name": "py-script", "version": "3.12.0", "class": "code", "command": ["/opt/code/py/bin/run"]}
],
"run_as": {"uid": 2001, "gid": 2001},
"allow_same_user": false,
@@@ -36,10 +37,12 @@ the server. The worker is **off by default**; off, the Node refuses every job `c
}
```
--- `functions`: `name` and `version` match a job's `artifact` exactly (the contract's patterns; listed once each);
-- `command[0]` is an absolute path. `env` is added to the job's environment. A per-entry `class` (`function`, the
-- default, or `code`) is planned and not accepted yet: the config loader refuses unknown keys, so leave it out
-- until the `code` class ships; only `function` is advertised in `status.capabilities.worker` today. `artifact_dir` (default: the directory
++- `functions`: `name` and `version` match a job's `artifact` exactly (the contract's patterns; listed once per class);
++ `command[0]` is an absolute path. `env` is added to the job's environment. `class` is the runtime class the entry
++ implements: `function` (the default) or `code`; `browser`, `linux`, `desktop` and `gpu` are refused at load. A job
++ runs only an entry of its own class: a `code` job naming a `function` entry is an `unknown artifact`, and the
++ reverse. A `code` entry is a local declared runtime like a function (nothing is fetched); its job gets exactly the
++ same sandbox, limits and checks as a function job, never weaker. `artifact_dir` (default: the directory
of `command[0]`) is the one host directory a job gets besides the system paths, as a **copy** made as the job
starts, read-only, at the same path: nothing the host adds to it or changes in it later reaches the job (a Unix
socket or a FIFO that appears there, say, through which a job could reach a host process). The Node opens the
@@@ -87,9 +90,9 @@ the server. The worker is **off by default**; off, the Node refuses every job `c
not built yet (follow-up): with either, the probe fails and the worker stays off, never a silent `none` nor an
open network.
--The class `function` — advertised as the reserved Fabric capability `worker:function` in
--`status.capabilities.worker` (the `platform.resource-offer@1` namespace OpenVibe.Run routes on) — is offered only when `enabled` is true, the OS is Linux
--(amd64 or arm64: the architectures with a seccomp allowlist), at least one `functions` entry is declared and a **boot-time probe** passes: it starts `/bin/sh`
++The classes `function` and `code` — advertised as the reserved Fabric capabilities `worker:function` and `worker:code` in
++`status.capabilities.worker` (the `platform.resource-offer@1` namespace OpenVibe.Run routes on) — are each offered only when `enabled` is true, the OS is Linux
++(amd64 or arm64: the architectures with a seccomp allowlist), at least one `functions` entry of that class is declared and a **boot-time probe** passes: it starts `/bin/sh`
the way a job is started and checks every control from the host: the sandbox was set up; the process has user,
mount, network, PID, IPC, UTS and cgroup namespaces of its own, `NoNewPrivs` and a seccomp filter; nothing in its
root but `/tmp`, `/proc` and the device nodes is mounted writable; it is in its own cgroup with every controller
@@@ -214,10 +217,10 @@ Refused with `nack` (keyed by the job id; the first matching row wins, after the
| `message` | `fault_code` | when |
|---------------------------------------|----------------------------|---------------------------------------------------------|
--| `class not available` | `unsupported` | the worker is off (disabled, not Linux, probe failed), or the class is not `function` |
++| `class not available` | `unsupported` | the worker is off (disabled, not Linux, probe failed), or no declared entry has the job's class (`function` or `code`) |
| `the e-stop or local stop is latched` | `estopped` or `local_stop` | the stop latch is set |
| `class not available` | `shutting_down` | the Node is stopping |
--| `unknown artifact` | `unsupported` | no declared function has that name and exact version |
++| `unknown artifact` | `unsupported` | no declared entry of the job's class has that name and exact version |
| `worker busy` | `not_ready` | `max_jobs` jobs are running |
Accepted jobs end with `job_exit.reason`:
ddiff --git a/internal/config/config.go b/internal/config/config.go
iindex 8bee62a..bfff9b4 100644
---- a/internal/config/config.go
++++ b/internal/config/config.go
@@@ -107,12 +107,12 @@ type Config struct {
// Python is the interpreter for plugins without an explicit command (default: <state>/venv/bin/python, then python3).
Python string `json:"python,omitempty"`
LogLevel string `json:"log_level,omitempty"`
-- // Worker runs `function` jobs from the control link (docs/worker.md). Off unless enabled.
++ // Worker runs `function` and `code` jobs from the control link (docs/worker.md). Off unless enabled.
Worker WorkerConfig `json:"worker,omitzero"`
}
// WorkerConfig is the job worker. Disabled (the default), every job is refused "class not available". Only the
--// functions listed here ever run: a job names one by name and exact version, and nothing is downloaded.
++// entries listed here ever run: a job names one by name, exact version and class, and nothing is downloaded.
type WorkerConfig struct {
Enabled bool `json:"enabled,omitempty"`
Functions []FunctionConfig `json:"functions,omitempty"`
@@@ -138,10 +138,13 @@ const (
EgressOpenVibeOnly = "openvibe-only"
)
--// FunctionConfig is one function a job may run: Command is started with the job's args as JSON on stdin.
++// FunctionConfig is one artifact a job may run: Command is started with the job's args as JSON on stdin.
type FunctionConfig struct {
-- Name string `json:"name"`
-- Version string `json:"version"`
++ Name string `json:"name"`
++ Version string `json:"version"`
++ // Class is the runtime class this entry implements: function (the default) or code. A job runs it only when its
++ // class matches; browser, linux, desktop and gpu are refused at load.
++ Class string `json:"class,omitempty"`
Command []string `json:"command"` // Command[0] is an absolute path
Env map[string]string `json:"env,omitempty"`
// ArtifactDir is the directory each job gets a read-only copy of, made as it starts, in its private root (default:
@@@ -149,6 +152,20 @@ type FunctionConfig struct {
ArtifactDir string `json:"artifact_dir,omitempty"`
}
++// The classes a declared entry may implement.
++const (
++ ClassFunction = "function"
++ ClassCode = "code"
++)
++
++// EffectiveClass is the entry's runtime class, function when none is set.
++func (f FunctionConfig) EffectiveClass() string {
++ if f.Class == "" {
++ return ClassFunction
++ }
++ return f.Class
++}
++
// Artifact is the directory a job of the function gets a read-only copy of.
func (f FunctionConfig) Artifact() string {
if f.ArtifactDir != "" {
@@@ -355,10 +372,14 @@ func (w WorkerConfig) validate() error {
if !functionNameRe.MatchString(f.Name) || !functionVersionRe.MatchString(f.Version) {
return fmt.Errorf("config: worker function %q version %q: name or version is not a valid artifact name or exact version", f.Name, f.Version)
}
-- if seen[f.Name+"@"+f.Version] {
-- return fmt.Errorf("config: worker function %s@%s listed twice", f.Name, f.Version)
++ if c := f.EffectiveClass(); c != ClassFunction && c != ClassCode {
++ return fmt.Errorf("config: worker function %s@%s has class %q: only function or code are implemented", f.Name, f.Version, f.Class)
++ }
++ key := f.Name + "@" + f.Version + "@" + f.EffectiveClass()
++ if seen[key] {
++ return fmt.Errorf("config: worker function %s@%s (class %s) listed twice", f.Name, f.Version, f.EffectiveClass())
}
-- seen[f.Name+"@"+f.Version] = true
++ seen[key] = true
if len(f.Command) == 0 || !filepath.IsAbs(f.Command[0]) {
return fmt.Errorf("config: worker function %s@%s needs a command starting with an absolute path", f.Name, f.Version)
}
ddiff --git a/internal/config/config_test.go b/internal/config/config_test.go
iindex 5c143e8..ae413c4 100644
---- a/internal/config/config_test.go
++++ b/internal/config/config_test.go
@@@ -86,3 +86,38 @@ func TestWorkerNameReserved(t *testing.T) {
t.Fatal("a plugin named worker was accepted")
}
}
++
++// TestWorkerEntryClass: an entry is a function unless it says code; the other runtime classes are refused at load,
++// and one name@version may be declared once per class.
++func TestWorkerEntryClass(t *testing.T) {
++ run := filepath.Join(t.TempDir(), "run") // absolute on every platform: Validate wants an absolute command
++ entry := func(class string) FunctionConfig {
++ return FunctionConfig{Name: "thumbnail", Version: "1.2.0", Class: class, Command: []string{run}}
++ }
++ if c := entry("").EffectiveClass(); c != ClassFunction {
++ t.Fatalf("no class is %q", c)
++ }
++ for _, class := range []string{"", "function", "code"} {
++ c := Default()
++ c.Worker.Functions = []FunctionConfig{entry(class)}
++ if err := c.Validate(); err != nil {
++ t.Fatalf("class %q: %v", class, err)
++ }
++ }
++ for _, class := range []string{"browser", "linux", "desktop", "gpu", "wasm", "Code"} {
++ c := Default()
++ c.Worker.Functions = []FunctionConfig{entry(class)}
++ if err := c.Validate(); err == nil {
++ t.Fatalf("class %q was accepted", class)
++ }
++ }
++ c := Default()
++ c.Worker.Functions = []FunctionConfig{entry(""), entry("code")}
++ if err := c.Validate(); err != nil {
++ t.Fatalf("one artifact as function and code: %v", err)
++ }
++ c.Worker.Functions = append(c.Worker.Functions, entry("function"))
++ if err := c.Validate(); err == nil {
++ t.Fatal("a function entry with no class and one with class function were both accepted")
++ }
++}
ddiff --git a/internal/node/node.go b/internal/node/node.go
iindex d9a389c..140bcb0 100644
---- a/internal/node/node.go
++++ b/internal/node/node.go
@@@ -159,10 +159,8 @@ func New(opt Options) (*Node, error) {
n.worker = w
w.SetStopped(stopFault(n.latch.State()))
n.jobs.admit = w.Admit
-- // Advertise function jobs only when the worker has a function to run.
-- if len(opt.Config.Worker.Functions) > 0 {
-- n.jobs.classes = []string{protocol.ClassFunction}
-- }
++ // Advertise a class only when an entry implements it, and only here, after the sandbox probe passed.
++ n.jobs.classes = w.Classes()
opt.Log.Info("worker on", "functions", len(opt.Config.Worker.Functions))
}
}
ddiff --git a/internal/node/worker_test.go b/internal/node/worker_test.go
iindex bf0142b..59dc49e 100644
---- a/internal/node/worker_test.go
++++ b/internal/node/worker_test.go
@@@ -55,6 +55,12 @@ func TestWorkerHelperProcess(t *testing.T) {
// startWorkerNode pairs a Node with no plugin and no video whose worker runs the `hello` function, and waits for the
// status that advertises it. It is skipped where the kernel does not let this user create the job namespaces.
func startWorkerNode(t *testing.T) *env {
++ t.Helper()
++ return startHelloNode(t, helloWorker(t))
++}
++
++// helloWorker is a worker config whose one function, [email protected], runs this test binary's TestWorkerHelperProcess.
++func helloWorker(t *testing.T) config.WorkerConfig {
t.Helper()
exe, err := os.Executable()
if err != nil {
@@@ -69,9 +75,21 @@ func startWorkerNode(t *testing.T) *env {
if os.Geteuid() == 0 { // as root, jobs run as nobody: the test binary must then lie in a directory nobody can read
cfg.AllowSameUser, cfg.RunAs = false, &config.RunAs{UID: 65534, GID: 65534}
}
++ return cfg
++}
++
++// startHelloNode starts a worker Node with cfg and checks its status advertises the classes its entries implement.
++func startHelloNode(t *testing.T, cfg config.WorkerConfig) *env {
++ t.Helper()
e, status := startWorkerNodeWith(t, cfg, true)
caps := cfg.Caps.WithDefaults()
-- want := map[string]any{"capabilities": []any{"worker:function"}, "runtime_classes": []any{protocol.ClassFunction}, "max_jobs": float64(caps.MaxJobs),
++ var names, classes []any
++ for _, c := range []string{protocol.ClassFunction, protocol.ClassCode} {
++ if slices.ContainsFunc(cfg.Functions, func(f config.FunctionConfig) bool { return f.EffectiveClass() == c }) {
++ names, classes = append(names, "worker:"+c), append(classes, c)
++ }
++ }
++ want := map[string]any{"capabilities": names, "runtime_classes": classes, "max_jobs": float64(caps.MaxJobs),
"max_ttl_ms": float64(caps.MaxTTLMS), "max_wall_ms": float64(caps.MaxWallMS),
"max_cpu_ms": float64(caps.MaxCPUMS), "max_mem_bytes": float64(caps.MaxMemBytes),
"max_output_bytes": float64(caps.MaxOutputBytes)}
@@@ -149,7 +167,8 @@ func TestWorkerWithNoFunctions(t *testing.T) {
func TestWorkerFailedProbeDoesNotAdvertiseFunctions(t *testing.T) {
e, status := startWorkerNodeWith(t, config.WorkerConfig{Enabled: true,
-- Functions: []config.FunctionConfig{{Name: "hello", Version: "1.0.0", Command: []string{"/bin/true"}}}}, false)
++ Functions: []config.FunctionConfig{{Name: "hello", Version: "1.0.0", Command: []string{"/bin/true"}},
++ {Name: "script", Version: "1.0.0", Class: config.ClassCode, Command: []string{"/bin/true"}}}}, false)
if classes := e.node.jobs.runtimeClasses(); len(classes) != 0 {
t.Fatalf("runtime classes after failed probe = %v, want none", classes)
}
@@@ -161,6 +180,57 @@ func TestWorkerFailedProbeDoesNotAdvertiseFunctions(t *testing.T) {
}
}
++// TestCodeJobRunsOverTheLink: a declared code entry advertises worker:code next to worker:function, and a code job
++// runs it in the same sandbox, while a code job naming the function entry is an unknown artifact.
++func TestCodeJobRunsOverTheLink(t *testing.T) {
++ cfg := helloWorker(t)
++ script := cfg.Functions[0]
++ script.Name, script.Class = "script", config.ClassCode
++ cfg.Functions = append(cfg.Functions, script)
++ e := startHelloNode(t, cfg)
++ if c := e.node.jobs.runtimeClasses(); !slices.Equal(c, []string{protocol.ClassFunction, protocol.ClassCode}) {
++ t.Fatalf("runtime classes %v", c)
++ }
++
++ j := testJob(jobID(1))
++ j.Class, j.Artifact = protocol.ClassCode, &protocol.Artifact{Name: "hello", Version: "1.0.0"}
++ if nk := e.jobNack(j, protocol.JobUnknownArtifact); nk.FaultCode != protocol.FaultUnsupported {
++ t.Fatalf("code job naming a function entry: %+v", nk)
++ }
++
++ j = testJob(jobID(2))
++ j.Class, j.Artifact = protocol.ClassCode, &protocol.Artifact{Name: "script", Version: "1.0.0"}
++ j.Args = json.RawMessage(fmt.Sprintf(`{"marker":%q,"hold_ms":1}`, filepath.Join(e.paths.StateDir, "starts")))
++ j.Limits.MemBytes = 4 << 30
++ if r := e.jobReply(j); r != (protocol.Ack{ID: j.ID}) {
++ t.Fatalf("%+v", r)
++ }
++ f, err := e.conn.Expect(protocol.TypeJobExit, 15*time.Second, func(m protocol.Message) bool {
++ ex, ok := m.(protocol.JobExit)
++ return ok && ex.ID == j.ID
++ })
++ if err != nil {
++ t.Fatal(err)
++ }
++ if ex := f.Msg.(protocol.JobExit); ex.Reason != protocol.ExitExited || ex.Code == nil || *ex.Code != 0 || string(ex.Result) != `{"ok":true}` {
++ t.Fatalf("%+v %s", ex, ex.Result)
++ }
++ if _, err := os.Stat(filepath.Join(e.paths.StateDir, "starts")); !errors.Is(err, fs.ErrNotExist) {
++ t.Fatalf("the code job wrote outside its sandbox: %v", err)
++ }
++}
++
++// TestCodeJobRefusedByFunctionWorker: a worker with only function entries does not advertise code, and a code job is
++// refused class not available before it reaches the worker.
++func TestCodeJobRefusedByFunctionWorker(t *testing.T) {
++ e := startWorkerNode(t)
++ j := testJob(jobID(1))
++ j.Class, j.Artifact = protocol.ClassCode, &protocol.Artifact{Name: "hello", Version: "1.0.0"}
++ if nk := e.jobNack(j, protocol.JobNotAvailable); nk.FaultCode != protocol.FaultUnsupported {
++ t.Fatalf("code job on a function-only worker: %+v", nk)
++ }
++}
++
// TestJobRunsOverTheLink: a job sent by the fake Bot is acked, then job_started, job_stdout, job_usage (second 0)
// and job_exit arrive in that order; a job frame resent while it runs is acked again and starts no second process;
// after the end a resent job is answered with the same job_exit, before and after job_exit_ack.
ddiff --git a/internal/protocol/protocol.go b/internal/protocol/protocol.go
iindex 0b375c1..ff25479 100644
---- a/internal/protocol/protocol.go
++++ b/internal/protocol/protocol.go
@@@ -107,7 +107,7 @@ const (
JobBadID = "missing or malformed job id"
JobBadFrame = "malformed job"
JobUnknownClass = "unknown class"
-- JobNoArtifact = "function job needs an artifact with an exact version"
++ JobNoArtifact = "function or code job needs an artifact with an exact version"
JobNetUnsupported = "net policy not supported"
JobBadLimits = "missing or invalid args, ttl_ms or limits"
JobNotAvailable = "class not available"
@@@ -331,7 +331,7 @@ func (r JobRequest) Check(available []string) (fault, reason string) {
return FaultBadFrame, JobBadFrame
case !contains(RuntimeClasses, j.Class):
return FaultUnsupported, JobUnknownClass
-- case j.Class == ClassFunction && (j.Artifact == nil || !artifactRe.MatchString(j.Artifact.Name) || !artifactVerRe.MatchString(j.Artifact.Version)):
++ case (j.Class == ClassFunction || j.Class == ClassCode) && (j.Artifact == nil || !artifactRe.MatchString(j.Artifact.Name) || !artifactVerRe.MatchString(j.Artifact.Version)):
return FaultBadValue, JobNoArtifact
case j.Net != "" && j.Net != NetDeny:
return FaultUnsupported, JobNetUnsupported
ddiff --git a/internal/protocol/protocol_test.go b/internal/protocol/protocol_test.go
iindex 59a20d6..b4811c3 100644
---- a/internal/protocol/protocol_test.go
++++ b/internal/protocol/protocol_test.go
@@@ -279,7 +279,11 @@ func TestJobCheck(t *testing.T) {
{"net deny taken", func(j *Job) { j.Net = NetDeny }, []string{ClassFunction}, ""},
{"nothing advertised", func(*Job) {}, nil, JobNotAvailable},
{"other class advertised", func(*Job) {}, []string{ClassCode}, JobNotAvailable},
-- {"code needs no artifact", func(j *Job) { j.Class, j.Artifact = ClassCode, nil }, nil, JobNotAvailable},
++ {"code taken", func(j *Job) { j.Class = ClassCode }, []string{ClassFunction, ClassCode}, ""},
++ {"code not advertised", func(j *Job) { j.Class = ClassCode }, []string{ClassFunction}, JobNotAvailable},
++ {"code needs an artifact", func(j *Job) { j.Class, j.Artifact = ClassCode, nil }, []string{ClassCode}, JobNoArtifact},
++ {"code needs a version", func(j *Job) { j.Class, j.Artifact.Version = ClassCode, "^1" }, []string{ClassCode}, JobNoArtifact},
++ {"browser needs no artifact", func(j *Job) { j.Class, j.Artifact = ClassBrowser, nil }, nil, JobNotAvailable},
{"no id", func(j *Job) { j.ID = "" }, []string{ClassFunction}, JobBadID},
{"id without prefix", func(j *Job) { j.ID = "01JAB2C3D4E5F6G7H8J9K0MNPQ" }, []string{ClassFunction}, JobBadID},
{"id with lower case", func(j *Job) { j.ID = "job_01jab2c3d4e5f6g7h8j9k0mnpq" }, []string{ClassFunction}, JobBadID},
ddiff --git a/internal/worker/worker.go b/internal/worker/worker.go
iindex ee00149..68e1252 100644
---- a/internal/worker/worker.go
++++ b/internal/worker/worker.go
@@@ -103,10 +103,10 @@ func (w *Worker) Admit(j protocol.Job) (fault, reason string) {
if w.closed {
return protocol.FaultShuttingDown, protocol.JobNotAvailable
}
-- if j.Class != protocol.ClassFunction {
++ if j.Class != protocol.ClassFunction && j.Class != protocol.ClassCode {
return protocol.FaultUnsupported, protocol.JobNotAvailable
}
-- fn, ok := w.function(j.Artifact)
++ fn, ok := w.entry(j.Class, j.Artifact)
if !ok {
return protocol.FaultUnsupported, protocol.JobUnknownArtifact
}
@@@ -125,18 +125,34 @@ func (w *Worker) Admit(j protocol.Job) (fault, reason string) {
return "", ""
}
--func (w *Worker) function(a *protocol.Artifact) (config.FunctionConfig, bool) {
++// entry finds the declared artifact that implements class: a code job never runs a function entry, nor the reverse.
++func (w *Worker) entry(class string, a *protocol.Artifact) (config.FunctionConfig, bool) {
if a == nil {
return config.FunctionConfig{}, false
}
for _, f := range w.cfg.Functions {
-- if f.Name == a.Name && f.Version == a.Version {
++ if f.Name == a.Name && f.Version == a.Version && f.EffectiveClass() == class {
return f, true
}
}
return config.FunctionConfig{}, false
}
++// Classes is the runtime classes the declared entries implement, in the order of protocol.RuntimeClasses: none
++// when no entry is declared.
++func (w *Worker) Classes() []string {
++ var out []string
++ for _, c := range protocol.RuntimeClasses {
++ for _, f := range w.cfg.Functions {
++ if f.EffectiveClass() == c {
++ out = append(out, c)
++ break
++ }
++ }
++ }
++ return out
++}
++
// Launch runs an admitted job on its own goroutine. A job already launched, or unknown, is left alone.
func (w *Worker) Launch(id string) {
w.mu.Lock()
ddiff --git a/internal/worker/worker_linux.go b/internal/worker/worker_linux.go
iindex dbbf4eb..133bd7d 100644
---- a/internal/worker/worker_linux.go
++++ b/internal/worker/worker_linux.go
@@@ -280,7 +280,7 @@ func probe(cfg config.WorkerConfig, isolate isolateFunc) error {
err = scanArtifact(f.Artifact())
}
if err != nil {
-- return fmt.Errorf("function %s@%s: %w", f.Name, f.Version, err)
++ return fmt.Errorf("%s %s@%s: %w", f.EffectiveClass(), f.Name, f.Version, err)
}
}
root, err := os.MkdirTemp("", "openvibe-probe-")
ddiff --git a/internal/worker/worker_test.go b/internal/worker/worker_test.go
iindex 915b788..13fa6b3 100644
---- a/internal/worker/worker_test.go
++++ b/internal/worker/worker_test.go
@@@ -138,7 +138,7 @@ func TestAdmitRefusals(t *testing.T) {
}{
{testJob(1, "nope"), protocol.FaultUnsupported, protocol.JobUnknownArtifact},
{wrongVersion, protocol.FaultUnsupported, protocol.JobUnknownArtifact},
-- {code, protocol.FaultUnsupported, protocol.JobNotAvailable},
++ {code, protocol.FaultUnsupported, protocol.JobUnknownArtifact},
} {
if f, r := w.Admit(c.j); f != c.fault || r != c.reason {
t.Errorf("%s %+v: %q %q, want %q %q", c.j.Class, c.j.Artifact, f, r, c.fault, c.reason)
@@@ -176,6 +176,42 @@ func TestAdmitRefusals(t *testing.T) {
}
}
++// TestAdmitByClass: a job runs only an entry of its own class (a declared function is no code artifact, nor the
++// reverse), a class no entry implements is never available, and Classes lists the declared classes in protocol order.
++func TestAdmitByClass(t *testing.T) {
++ if c := New(helperConfig(t, config.WorkerCaps{}), (&sink{}).send, quiet()).Classes(); len(c) != 0 {
++ t.Fatalf("no entry declared, classes %v", c)
++ }
++ cfg := helperConfig(t, config.WorkerCaps{MaxJobs: 8}, "sleep")
++ script := cfg.Functions[0]
++ script.Name, script.Class = "script", config.ClassCode
++ cfg.Functions = append([]config.FunctionConfig{script}, cfg.Functions...)
++ w := New(cfg, (&sink{}).send, quiet())
++ if c := w.Classes(); len(c) != 2 || c[0] != protocol.ClassFunction || c[1] != protocol.ClassCode {
++ t.Fatalf("classes %v", c)
++ }
++ job := func(n int, class, name string) protocol.Job {
++ j := testJob(n, name)
++ j.Class = class
++ return j
++ }
++ for _, c := range []struct {
++ j protocol.Job
++ fault, reason string
++ }{
++ {job(1, protocol.ClassCode, "script"), "", ""},
++ {job(2, protocol.ClassFunction, "sleep"), "", ""},
++ {job(3, protocol.ClassFunction, "script"), protocol.FaultUnsupported, protocol.JobUnknownArtifact},
++ {job(4, protocol.ClassCode, "sleep"), protocol.FaultUnsupported, protocol.JobUnknownArtifact},
++ {job(5, protocol.ClassBrowser, "script"), protocol.FaultUnsupported, protocol.JobNotAvailable},
++ } {
++ if f, r := w.Admit(c.j); f != c.fault || r != c.reason {
++ t.Errorf("%s %+v: %q %q, want %q %q", c.j.Class, c.j.Artifact, f, r, c.fault, c.reason)
++ }
++ }
++ w.Close()
++}
++
// TestIsolationUnavailableRefused: when the namespaces cannot be created the probe fails (the class is not
// advertised), and a job whose isolation fails ends `failed` without its process ever starting.
func TestIsolationUnavailableRefused(t *testing.T) {