openvibers/openvibe.link Commits
Answer control-link job frames and advertise worker runtime classes (#11)
Changes
@@@ -178,6 +178,41 @@ What the Node does with a command, in order: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.++**Running jobs is not implemented yet:** the Node advertises no runtime class, so it 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_cancel` | `id` |++| server → device | `job_exit_ack` | `id` |++| device → server | `job_started` | `id`, `started_ms` |++| device → server | `job_stdout` | `id`, `chunk_seq` (the job's own counter from 1; the envelope owns `seq`), `chunk` |++| device → server | `job_usage` | `id`, `started_ms`, `second`, `cpu_ms` |++| device → server | `job_exit` | `id`, `reason` (`exited`, `cancelled`, `ttl`, `limit`, `stopped`, `failed`), `code`, `result`, `usage` {`started_ms`, `wall_ms`, `cpu_ms`, `mem_peak_bytes`} |++++A `job` is answered `ack` (accepted) or `nack` keyed by the job id. A nack's `message` names the first problem found,++checked in this order:++++| `message` | `fault_code` | when |++|----------------------------------------------|---------------|-------------------------------------------------------------|++| `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) |++| `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 (today: every class) |++++A resent `job` with an id already answered gets the same answer again, never a second acceptance; the Node++remembers the last 1024 well-formed ids. `job_cancel` for an id that is not running (unknown, refused or ended) and++`job_exit_ack` for an unknown id are ignored, with no answer: an `ack` keyed by a job id means the job was accepted.++A Node that runs jobs lists its classes in `status.capabilities.worker` as `{"runtime_classes": ["function"]}`; the++key is absent while it runs none, and `worker` is reserved as a plugin name.++### Fault codes`bad_frame`, `bad_value`, `unsupported`, `not_allowed`, `estopped`, `local_stop`, `expired`, `no_heartbeat`,@@@ -212,6 +212,9 @@ func (c *Config) Validate() error {if p.Name == "" || strings.ContainsAny(p.Name, " /\\") {return fmt.Errorf("config: plugin name %q is empty or has spaces or slashes", p.Name)}++ if p.Name == "worker" { // status.capabilities.worker holds the job runtime classes++ return fmt.Errorf("config: plugin name %q is reserved", p.Name)++ }if seen[p.Name] {return fmt.Errorf("config: plugin %q listed twice", p.Name)}@@@ -65,3 +65,12 @@ func TestPluginCommand(t *testing.T) {t.Fatalf("%v", got)}}++++// TestWorkerNameReserved: status.capabilities.worker holds the job runtime classes, so no plugin may be named worker.++func TestWorkerNameReserved(t *testing.T) {++ c := Default()++ c.Plugins = append(c.Plugins, PluginConfig{Name: "worker", Command: []string{"/bin/true"}})++ if err := c.Validate(); err == nil {++ t.Fatal("a plugin named worker was accepted")++ }++}@@@ -364,6 +364,15 @@ func (c *Conn) SendError(code, detail string) error {return c.Send(protocol.Error{Code: code, Detail: detail})}++// Job sends a job frame; the device answers ack or nack keyed by j.ID (see Reply).++func (c *Conn) Job(j protocol.Job) error { return c.Send(protocol.JobRequest{Job: j}) }++++// JobCancel asks the device to stop job id.++func (c *Conn) JobCancel(id string) error { return c.Send(protocol.JobCancel{ID: id}) }++++// JobExitAck tells the device job id's job_exit is recorded.++func (c *Conn) JobExitAck(id string) error { return c.Send(protocol.JobExitAck{ID: id}) }++// setEstop records the robot's latch: Bot keeps it from the estop frames it sends and the latched:true estop_state// frames it receives.func (c *Conn) setEstop(latched bool) {nnew file mode 100644@@@ -0,0 +1,85 @@++package node++++import (++ "sync"++++ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"++)++++// jobSeenMax bounds how many job ids the Node remembers answers for.++const jobSeenMax = 1024++++// jobs answers job frames. Nothing runs yet: a job is refused unless its class is advertised, and an accepted job is++// only remembered (the worker that runs it fills classes and executes accepted jobs).++type jobs struct {++ mu sync.Mutex++ classes []string // runtime classes this Node runs and advertises; empty: none++ answers map[string]protocol.Message // job id → the ack or nack it got, resent unchanged for a repeated job++ order []string // ids in answers, oldest first++}++++func newJobs(classes []string) *jobs {++ return &jobs{classes: append([]string(nil), classes...), answers: map[string]protocol.Message{}}++}++++func (j *jobs) runtimeClasses() []string {++ j.mu.Lock()++ defer j.mu.Unlock()++ return append([]string(nil), j.classes...)++}++++// request answers a job frame. A job id already answered gets the same answer again, never a second acceptance;++// a job whose id is malformed is nacked but not remembered, so junk cannot evict real ids.++func (j *jobs) request(r protocol.JobRequest) (answer protocol.Message, accepted bool) {++ j.mu.Lock()++ defer j.mu.Unlock()++ id := r.Job.ID++ if a, ok := j.answers[id]; ok {++ return a, false++ }++ fault, reason := r.Check(j.classes)++ if fault != "" {++ answer = protocol.Nack{ID: id, FaultCode: fault, Message: reason}++ } else {++ answer, accepted = protocol.Ack{ID: id}, true++ }++ if !protocol.ValidJobID(id) {++ return answer, false++ }++ if len(j.order) >= jobSeenMax {++ delete(j.answers, j.order[0])++ j.order = j.order[1:]++ }++ j.answers[id] = answer++ j.order = append(j.order, id)++ return answer, accepted++}++++// accepted reports whether id was accepted (and so would be running).++func (j *jobs) accepted(id string) bool {++ j.mu.Lock()++ defer j.mu.Unlock()++ _, ok := j.answers[id].(protocol.Ack)++ return ok++}++++// job answers a job frame with ack or nack, keyed by the job id.++func (n *Node) job(r protocol.JobRequest) {++ answer, accepted := n.jobs.request(r)++ if nk, ok := answer.(protocol.Nack); ok {++ n.log.Info("job refused", "id", r.Job.ID, "class", r.Job.Class, "reason", nk.Message, "err", r.Err)++ } else if accepted {++ n.log.Info("job accepted", "id", r.Job.ID, "class", r.Job.Class)++ }++ n.send(answer)++}++++// jobCancel stops an accepted job. An unknown, refused or ended id is ignored, as platform.job-frame@1 says, and no++// frame answers it: an ack keyed by a job id means the job was accepted.++func (n *Node) jobCancel(c protocol.JobCancel) {++ if !n.jobs.accepted(c.ID) {++ n.log.Debug("job_cancel for a job that is not running ignored", "id", c.ID)++ return++ }++ n.log.Warn("job_cancel: job execution is not implemented, nothing to stop", "id", c.ID)++}nnew file mode 100644@@@ -0,0 +1,171 @@++//go:build !windows++++package node++++import (++ "encoding/json"++ "fmt"++ "testing"++ "time"++++ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"++)++++func jobID(n int) string { return fmt.Sprintf("job_01JAB2C3D4E5F6G7H8J9K%05d", n) }++++func testJob(id string) protocol.Job {++ return protocol.Job{ID: id, Class: protocol.ClassFunction, Artifact: &protocol.Artifact{Name: "thumbnail", Version: "1.2.0"},++ Args: json.RawMessage(`{"width":320}`), TTLMS: 60000, Limits: protocol.JobLimits{WallMS: 30000, CPUMS: 30000, MemBytes: 268435456}}++}++++func (e *env) jobReply(j protocol.Job) protocol.Message {++ e.t.Helper()++ if err := e.conn.Job(j); err != nil {++ e.t.Fatal(err)++ }++ r, err := e.conn.Reply(j.ID, 5*time.Second)++ if err != nil {++ e.t.Fatal(err)++ }++ return r++}++++func (e *env) jobNack(j protocol.Job, reason string) protocol.Nack {++ e.t.Helper()++ r := e.jobReply(j)++ nk, ok := r.(protocol.Nack)++ if !ok || nk.ID != j.ID || nk.Message != reason || nk.FaultCode == "" {++ e.t.Fatalf("job %s: %+v, want a nack %q", j.ID, r, reason)++ }++ return nk++}++++// TestJobsRefused: with no runtime class advertised (nothing executes yet) every job is nacked, with the first problem++// as the reason, a resent job id gets the same nack, and job_cancel, job_exit_ack and unknown frames change nothing.++func TestJobsRefused(t *testing.T) {++ e := start(t)++ for _, f := range e.conn.Frames() {++ if s, ok := f.Msg.(protocol.Status); ok {++ if _, ok := s.Capabilities[protocol.CapWorker]; ok {++ t.Fatalf("worker capability advertised with no runtime class: %+v", s.Capabilities)++ }++ }++ }++++ first := e.jobNack(testJob(jobID(1)), protocol.JobNotAvailable)++ if first.FaultCode != protocol.FaultUnsupported {++ t.Fatalf("%+v", first)++ }++ if again := e.jobNack(testJob(jobID(1)), protocol.JobNotAvailable); again != first {++ t.Fatalf("resent job answered %+v, first %+v", again, first)++ }++++ j := testJob(jobID(2))++ j.Class = "quantum"++ e.jobNack(j, protocol.JobUnknownClass)++++ j = testJob(jobID(3))++ j.Net = "allow"++ e.jobNack(j, protocol.JobNetUnsupported)++++ e.jobNack(testJob("job_123"), protocol.JobBadID)++++ j = testJob(jobID(4))++ j.Artifact.Version = ">=1.0"++ e.jobNack(j, protocol.JobNoArtifact)++++ // A body that does not decode is nacked by its id, not dropped.++ bad := fmt.Sprintf(`{"v":1,"seq":900,"ts":%d,"type":"job","job":{"id":%q,"class":"function","ttl_ms":"soon"}}`, time.Now().UnixMilli(), jobID(5))++ if err := e.conn.SendRaw([]byte(bad)); err != nil {++ t.Fatal(err)++ }++ if r, err := e.conn.Reply(jobID(5), 5*time.Second); err != nil || r.(protocol.Nack).Message != protocol.JobBadFrame {++ t.Fatalf("%+v %v", r, err)++ }++++ // An unknown frame type, job_cancel and job_exit_ack for unknown ids: ignored, nothing answers them, and the next job++ // is still answered.++ if err := e.conn.SendRaw([]byte(fmt.Sprintf(`{"v":1,"seq":901,"ts":%d,"type":"future","id":%q}`, time.Now().UnixMilli(), jobID(6)))); err != nil {++ t.Fatal(err)++ }++ if err := e.conn.JobCancel(jobID(6)); err != nil {++ t.Fatal(err)++ }++ if err := e.conn.JobCancel(jobID(1)); err != nil { // refused earlier: not running either++ t.Fatal(err)++ }++ if err := e.conn.JobExitAck(jobID(6)); err != nil {++ t.Fatal(err)++ }++ e.jobNack(testJob(jobID(7)), protocol.JobNotAvailable)++ nacks1 := 0++ for _, f := range e.conn.Frames() {++ switch m := f.Msg.(type) {++ case protocol.Ack:++ t.Fatalf("ack with nothing accepted: %+v", m)++ case protocol.Nack:++ if m.ID == jobID(6) {++ t.Fatalf("job_cancel or job_exit_ack answered: %+v", m)++ }++ if m.ID == jobID(1) {++ nacks1++++ }++ }++ }++ if nacks1 != 2 {++ t.Fatalf("job %s answered %d times, want 2 (one per job frame)", jobID(1), nacks1)++ }++}++++// TestJobAcceptedOnce: once a class is advertised a job is acked, a resent one is acked again (never a second++// acceptance), and status.capabilities.worker lists the classes.++func TestJobAcceptedOnce(t *testing.T) {++ e := start(t)++ e.node.jobs.mu.Lock()++ e.node.jobs.classes = []string{protocol.ClassFunction}++ e.node.jobs.mu.Unlock()++++ if r := e.jobReply(testJob(jobID(1))); r != (protocol.Ack{ID: jobID(1)}) {++ t.Fatalf("%+v", r)++ }++ if _, accepted := e.node.jobs.request(protocol.JobRequest{Job: testJob(jobID(1))}); accepted {++ t.Fatal("a resent job was accepted twice")++ }++ if r := e.jobReply(testJob(jobID(1))); r != (protocol.Ack{ID: jobID(1)}) {++ t.Fatalf("resent: %+v", r)++ }++ if err := e.conn.JobCancel(jobID(1)); err != nil {++ t.Fatal(err)++ }++++ // The next connection's status advertises the class.++ e.conn.Close()++ e.conn = e.nextConn()++ _, err := e.conn.Expect(protocol.TypeStatus, 10*time.Second, func(m protocol.Message) bool {++ w, _ := m.(protocol.Status).Capabilities[protocol.CapWorker].(map[string]any)++ c, _ := w["runtime_classes"].([]any)++ return len(c) == 1 && c[0] == protocol.ClassFunction++ })++ if err != nil {++ t.Fatal(err)++ }++}++++// TestJobAnswersBounded: the Node remembers the answers of the last jobSeenMax valid job ids only, and never a++// malformed one.++func TestJobAnswersBounded(t *testing.T) {++ j := newJobs(nil)++ for i := 0; i <= jobSeenMax; i++ {++ j.request(protocol.JobRequest{Job: testJob(jobID(i))})++ }++ j.request(protocol.JobRequest{Job: testJob("job_bad")})++ if len(j.answers) != jobSeenMax || len(j.order) != jobSeenMax {++ t.Fatalf("%d answers, %d ids", len(j.answers), len(j.order))++ }++ if _, ok := j.answers[jobID(0)]; ok {++ t.Fatal("the oldest id was not forgotten")++ }++ if _, ok := j.answers[jobID(jobSeenMax)]; !ok {++ t.Fatal("the newest id was forgotten")++ }++}@@@ -42,6 +42,10 @@ type Options struct {Log *slog.LoggerVersion string++ // RuntimeClasses are the job classes this Node runs and advertises in status.capabilities.worker. None yet:++ // every job is refused "class not available" until the worker that executes jobs fills this in.++ RuntimeClasses []string++// For tests.HeartbeatInterval time.DurationLinkBackoffMin time.Duration@@@ -55,6 +59,7 @@ type Node struct {log *slog.Loggerlatch *safety.Latchdedup *safety.Dedup++ jobs *jobsmgr *plugins.Managerlink *link.Linkpub *video.Publisher@@@ -93,7 +98,7 @@ func New(opt Options) (*Node, error) {opt.Log.Error("latch", "err", err)}n := &Node{opt: opt, log: opt.Log, latch: latch, dedup: safety.NewDedup(4096, 10*time.Minute),-- telemetry: map[string]map[string]any{}, telemMS: 500}++ jobs: newJobs(opt.RuntimeClasses), telemetry: map[string]map[string]any{}, telemMS: 500}n.limits = safety.Merge(protocol.Limits{}, opt.Config.Limits.MaxSpeed, opt.Config.Limits.MaxTurn, opt.Config.Limits.MaxCommandMS)var specs []plugins.Spec@@@ -226,6 +231,13 @@ func (n *Node) Frame(f protocol.Frame) {n.remoteClear(m.By)}n.sendEstopState()++ case protocol.JobRequest:++ n.job(m)++ case protocol.JobCancel:++ n.jobCancel(m)++ case protocol.JobExitAck:++ // No job runs yet, so no job_exit is ever waiting for this.++ n.log.Debug("job_exit_ack ignored", "id", m.ID)}}@@@ -772,6 +784,9 @@ func (n *Node) sendStatus() {s.Faults = append(s.Faults, *i.Fault)}}++ if c := n.jobs.runtimeClasses(); len(c) > 0 {++ s.Capabilities[protocol.CapWorker] = protocol.WorkerCapabilities{RuntimeClasses: c}++ }if n.pub != nil {s.Video = n.pub.State()}@@@ -11,6 +11,7 @@ import ("encoding/json""errors""fmt"++ "regexp")// Version is the envelope version, the `v` field of every frame.@@@ -55,6 +56,59 @@ const (TypeEstopState = "estop_state")++// Job frames (OpenVibe.Contracts platform.job-frame@1). Server → device: job, job_cancel, job_exit_ack. Device →++// server: job_started, job_stdout, job_usage, job_exit. A job is accepted or refused with ack/nack keyed by its id.++const (++ TypeJob = "job"++ TypeJobCancel = "job_cancel"++ TypeJobExitAck = "job_exit_ack"++ TypeJobStarted = "job_started"++ TypeJobStdout = "job_stdout"++ TypeJobUsage = "job_usage"++ TypeJobExit = "job_exit"++)++++// Runtime classes (platform.runtime-class@1).++const (++ ClassFunction = "function"++ ClassCode = "code"++ ClassBrowser = "browser"++ ClassLinux = "linux"++ ClassDesktop = "desktop"++ ClassGPU = "gpu"++)++++// RuntimeClasses lists every class the contract names; a worker runs only those it advertises.++var RuntimeClasses = []string{ClassFunction, ClassCode, ClassBrowser, ClassLinux, ClassDesktop, ClassGPU}++++// NetDeny is the only job network policy (and the default when net is absent).++const NetDeny = "deny"++++// CapWorker is the status.capabilities key a worker advertises its runtime classes under (WorkerCapabilities). It is++// present only when the Node runs at least one class.++const CapWorker = "worker"++++// Job refusal reasons, carried in nack.message (nack.fault_code is bad_value, bad_frame or unsupported).++const (++ JobBadID = "missing or malformed job id"++ JobBadFrame = "malformed job"++ JobUnknownClass = "unknown class"++ JobNoArtifact = "function 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"++)++++// job_exit reasons.++const (++ ExitExited = "exited"++ ExitCancelled = "cancelled"++ ExitTTL = "ttl"++ ExitLimit = "limit"++ ExitStopped = "stopped"++ ExitFailed = "failed"++)++// Command kinds.const (KindDrive = "drive"@@@ -185,8 +239,160 @@ type Error struct {Detail string `json:"detail,omitempty"`}++// JobLimits are the caps the worker enforces on a running job.++type JobLimits struct {++ WallMS int64 `json:"wall_ms"`++ CPUMS int64 `json:"cpu_ms"`++ MemBytes int64 `json:"mem_bytes"`++}++++// Artifact is the pre-registered artifact a function job runs, at one exact version.++type Artifact struct {++ Name string `json:"name"`++ Version string `json:"version"`++}++++// Job is platform.job@1.++type Job struct {++ ID string `json:"id"` // job_<ULID>, minted by the dispatcher: the idempotency and ack key++ Class string `json:"class"`++ Artifact *Artifact `json:"artifact,omitempty"`++ Args json.RawMessage `json:"args"`++ TTLMS int64 `json:"ttl_ms"`++ Limits JobLimits `json:"limits"`++ Net string `json:"net,omitempty"` // absent means deny++}++++// JobRequest is the `job` frame: run Job. A body that does not decode still yields the frame, with Err set and the++// id when it is a string, so the Node can nack it rather than drop it (the server resends a job until it is answered).++type JobRequest struct {++ Job Job `json:"job"`++ Err error `json:"-"`++}++++func (r *JobRequest) UnmarshalJSON(b []byte) error {++ var outer struct {++ Job json.RawMessage `json:"job"`++ }++ if err := json.Unmarshal(b, &outer); err != nil {++ return err++ }++ *r = JobRequest{}++ if len(outer.Job) == 0 {++ return nil++ }++ if err := json.Unmarshal(outer.Job, &r.Job); err != nil {++ var id struct {++ ID any `json:"id"`++ }++ _ = json.Unmarshal(outer.Job, &id)++ s, _ := id.ID.(string)++ r.Job, r.Err = Job{ID: s}, err++ }++ return nil++}++++var (++ jobIDRe = regexp.MustCompile(`^job_[0-9A-HJKMNP-TV-Z]{26}$`)++ artifactRe = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]{0,127}$`)++ artifactVerRe = regexp.MustCompile(`^[0-9A-Za-z][0-9A-Za-z.+-]{0,63}$`)++)++++// ValidJobID reports whether id is a job_<ULID>.++func ValidJobID(id string) bool { return jobIDRe.MatchString(id) }++++// Check returns the fault code and reason to nack this job with, or two empty strings when a worker running the++// classes in available takes it. The checks run in this order, so the reason names the first problem.++func (r JobRequest) Check(available []string) (fault, reason string) {++ j := r.Job++ switch {++ case !ValidJobID(j.ID):++ return FaultBadValue, JobBadID++ case r.Err != nil:++ 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)):++ return FaultBadValue, JobNoArtifact++ case j.Net != "" && j.Net != NetDeny:++ return FaultUnsupported, JobNetUnsupported++ case !isObject(j.Args) || j.TTLMS < 1 || j.Limits.WallMS < 1 || j.Limits.CPUMS < 1 || j.Limits.MemBytes < 1:++ return FaultBadValue, JobBadLimits++ case !contains(available, j.Class):++ return FaultUnsupported, JobNotAvailable++ }++ return "", ""++}++++func contains(list []string, s string) bool {++ for _, v := range list {++ if v == s {++ return true++ }++ }++ return false++}++++func isObject(b json.RawMessage) bool {++ b = bytes.TrimSpace(b)++ return len(b) >= 2 && b[0] == '{'++}++++// JobCancel stops a job; an unknown or ended id is ignored.++type JobCancel struct {++ ID string `json:"id"`++}++++// JobExitAck tells the worker every usage reading of the job is recorded, so it stops resending job_exit.++type JobExitAck struct {++ ID string `json:"id"`++}++// ---- device → server ----++// JobStarted: the job's process started at StartedMS (worker wall clock, Unix ms), which anchors every usage second.++type JobStarted struct {++ ID string `json:"id"`++ StartedMS int64 `json:"started_ms"`++}++++// JobStdout is a chunk of the job's stdout. ChunkSeq is the job's own counter from 1 (the envelope owns seq).++type JobStdout struct {++ ID string `json:"id"`++ ChunkSeq uint64 `json:"chunk_seq"`++ Chunk string `json:"chunk"`++}++++// JobUsage: wall-clock second Second (from 0 at StartedMS) of the job has fully elapsed. CPUMS is informational.++type JobUsage struct {++ ID string `json:"id"`++ StartedMS int64 `json:"started_ms"`++ Second int64 `json:"second"`++ CPUMS *int64 `json:"cpu_ms,omitempty"`++}++++// JobExitUsage is a job_exit's authoritative usage; StartedMS is set whenever WallMS > 0.++type JobExitUsage struct {++ StartedMS *int64 `json:"started_ms,omitempty"`++ WallMS int64 `json:"wall_ms"`++ CPUMS *int64 `json:"cpu_ms,omitempty"`++ MemPeakBytes *int64 `json:"mem_peak_bytes,omitempty"`++}++++// JobExit: the job ended. Code is null when the process was killed or never ran; Result is null unless it exited 0.++type JobExit struct {++ ID string `json:"id"`++ Reason string `json:"reason"`++ Code *int `json:"code"`++ Result json.RawMessage `json:"result"`++ Usage JobExitUsage `json:"usage"`++}++++// WorkerCapabilities is status.capabilities[CapWorker]: the runtime classes this Node runs.++type WorkerCapabilities struct {++ RuntimeClasses []string `json:"runtime_classes"`++}++type DriverStatus struct {Name string `json:"name"`Driver string `json:"driver,omitempty"`@@@ -271,6 +477,13 @@ func (Ack) MessageType() string { return TypeAck }func (Nack) MessageType() string { return TypeNack }func (Heartbeat) MessageType() string { return TypeHeartbeat }func (EstopState) MessageType() string { return TypeEstopState }++func (JobRequest) MessageType() string { return TypeJob }++func (JobCancel) MessageType() string { return TypeJobCancel }++func (JobExitAck) MessageType() string { return TypeJobExitAck }++func (JobStarted) MessageType() string { return TypeJobStarted }++func (JobStdout) MessageType() string { return TypeJobStdout }++func (JobUsage) MessageType() string { return TypeJobUsage }++func (JobExit) MessageType() string { return TypeJobExit }var registry = map[string]func() Message{TypeHello: func() Message { return &Hello{} },@@@ -285,6 +498,13 @@ var registry = map[string]func() Message{TypeNack: func() Message { return &Nack{} },TypeHeartbeat: func() Message { return &Heartbeat{} },TypeEstopState: func() Message { return &EstopState{} },++ TypeJob: func() Message { return &JobRequest{} },++ TypeJobCancel: func() Message { return &JobCancel{} },++ TypeJobExitAck: func() Message { return &JobExitAck{} },++ TypeJobStarted: func() Message { return &JobStarted{} },++ TypeJobStdout: func() Message { return &JobStdout{} },++ TypeJobUsage: func() Message { return &JobUsage{} },++ TypeJobExit: func() Message { return &JobExit{} },}// ErrUnknownType is returned (wrapped) by Decode for a frame whose type is not in this version's message set.@@@ -363,6 +583,20 @@ func derefMessage(m Message) Message {return *vcase *EstopState:return *v++ case *JobRequest:++ return *v++ case *JobCancel:++ return *v++ case *JobExitAck:++ return *v++ case *JobStarted:++ return *v++ case *JobStdout:++ return *v++ case *JobUsage:++ return *v++ case *JobExit:++ return *v}return m}@@@ -5,6 +5,7 @@ import ("errors""os""path/filepath"++ "reflect""testing")@@@ -179,3 +180,135 @@ func TestDecodeRejects(t *testing.T) {t.Fatalf("want ErrUnknownType, got %v", err)}}++++// TestJobFrameFixtures decodes every valid platform.job-frame@1 fixture from OpenVibe.Contracts (testdata/contracts),++// checks the fields the Node reads, and encodes it back to the same JSON object.++func TestJobFrameFixtures(t *testing.T) {++ const id = "job_01JAB2C3D4E5F6G7H8J9K0MNPQ"++ for _, name := range []string{"job", "job-cancel", "job-exit-ack", "job-started", "job-stdout", "job-usage", "job-exit", "job-exit-never-started"} {++ b, err := os.ReadFile(filepath.Join("testdata", "contracts", name+".json"))++ if err != nil {++ t.Fatal(err)++ }++ f, err := Decode(b)++ if err != nil {++ t.Fatalf("%s: %v", name, err)++ }++ switch m := f.Msg.(type) {++ case JobRequest:++ j := m.Job++ if m.Err != nil || j.ID != id || j.Class != ClassFunction || j.Artifact == nil || j.Artifact.Name != "thumbnail" ||++ j.Artifact.Version != "1.2.0" || j.TTLMS != 60000 || j.Limits != (JobLimits{WallMS: 30000, CPUMS: 30000, MemBytes: 268435456}) || j.Net != "" {++ t.Errorf("%s: %+v", name, m)++ }++ if fault, reason := m.Check([]string{ClassFunction}); fault != "" {++ t.Errorf("%s: the contract's example is refused: %s %s", name, fault, reason)++ }++ case JobCancel:++ if m.ID != id {++ t.Errorf("%s: %+v", name, m)++ }++ case JobExitAck:++ if m.ID != id {++ t.Errorf("%s: %+v", name, m)++ }++ case JobStarted:++ if m.ID != id || m.StartedMS != 1790935200250 {++ t.Errorf("%s: %+v", name, m)++ }++ case JobStdout:++ if m.ID != id || m.ChunkSeq != 1 || f.Seq != 13 || m.Chunk != "resized 1 of 1\n" {++ t.Errorf("%s: %+v (envelope seq %d)", name, m, f.Seq)++ }++ case JobUsage:++ if m.ID != id || m.StartedMS != 1790935200250 || m.Second != 0 || m.CPUMS == nil || *m.CPUMS != 640 {++ t.Errorf("%s: %+v", name, m)++ }++ case JobExit:++ ran := name == "job-exit"++ if m.ID != id || (ran && (m.Reason != ExitExited || m.Code == nil || *m.Code != 0 || m.Usage.StartedMS == nil || m.Usage.WallMS != 2350)) ||++ (!ran && (m.Reason != ExitTTL || m.Code != nil || string(m.Result) != "null" || m.Usage.StartedMS != nil || m.Usage.WallMS != 0)) {++ t.Errorf("%s: %+v", name, m)++ }++ default:++ t.Fatalf("%s: decoded as %T", name, f.Msg)++ }++ out, err := Encode(f.Seq, f.TS, f.Msg)++ if err != nil {++ t.Fatal(err)++ }++ var want, got any++ _ = json.Unmarshal(b, &want)++ if err := json.Unmarshal(out, &got); err != nil {++ t.Fatal(err)++ }++ if !reflect.DeepEqual(want, got) {++ t.Errorf("%s does not round-trip:\n want %v\n got %v", name, want, got)++ }++ }++}++++func TestJobCheck(t *testing.T) {++ const id = "job_01JAB2C3D4E5F6G7H8J9K0MNPQ"++ valid := func() Job {++ return Job{ID: id, Class: ClassFunction, Artifact: &Artifact{Name: "thumbnail", Version: "1.2.0"}, Args: json.RawMessage(`{}`),++ TTLMS: 1000, Limits: JobLimits{WallMS: 1, CPUMS: 1, MemBytes: 1}}++ }++ cases := []struct {++ name string++ edit func(*Job)++ available []string++ reason string++ }{++ {"taken", func(*Job) {}, []string{ClassFunction}, ""},++ {"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},++ {"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},++ {"unknown class", func(j *Job) { j.Class = "quantum" }, []string{ClassFunction}, JobUnknownClass},++ {"no class", func(j *Job) { j.Class = "" }, []string{ClassFunction}, JobUnknownClass},++ {"no artifact", func(j *Job) { j.Artifact = nil }, []string{ClassFunction}, JobNoArtifact},++ {"no version", func(j *Job) { j.Artifact.Version = "" }, []string{ClassFunction}, JobNoArtifact},++ {"version range", func(j *Job) { j.Artifact.Version = "^1.2" }, []string{ClassFunction}, JobNoArtifact},++ {"version span", func(j *Job) { j.Artifact.Version = "1.2 - 2" }, []string{ClassFunction}, JobNoArtifact},++ {"bad artifact name", func(j *Job) { j.Artifact.Name = "Thumbnail" }, []string{ClassFunction}, JobNoArtifact},++ {"net allow", func(j *Job) { j.Net = "allow" }, []string{ClassFunction}, JobNetUnsupported},++ {"no args", func(j *Job) { j.Args = nil }, []string{ClassFunction}, JobBadLimits},++ {"args not an object", func(j *Job) { j.Args = json.RawMessage(`[1]`) }, []string{ClassFunction}, JobBadLimits},++ {"no ttl", func(j *Job) { j.TTLMS = 0 }, []string{ClassFunction}, JobBadLimits},++ {"no mem limit", func(j *Job) { j.Limits.MemBytes = 0 }, []string{ClassFunction}, JobBadLimits},++ }++ for _, c := range cases {++ j := valid()++ c.edit(&j)++ if _, reason := (JobRequest{Job: j}).Check(c.available); reason != c.reason {++ t.Errorf("%s: reason %q, want %q", c.name, reason, c.reason)++ }++ }++}++++// TestJobUndecodableBody: a job body of the wrong shape still decodes as a job frame, with the id and Err, so the++// Node nacks it instead of dropping it.++func TestJobUndecodableBody(t *testing.T) {++ f, err := Decode([]byte(`{"v":1,"seq":1,"ts":1,"type":"job","job":{"id":"job_01JAB2C3D4E5F6G7H8J9K0MNPQ","class":"function","ttl_ms":"soon"}}`))++ if err != nil {++ t.Fatal(err)++ }++ r := f.Msg.(JobRequest)++ if r.Err == nil || r.Job.ID != "job_01JAB2C3D4E5F6G7H8J9K0MNPQ" {++ t.Fatalf("%+v", r)++ }++ if fault, reason := r.Check([]string{ClassFunction}); fault != FaultBadFrame || reason != JobBadFrame {++ t.Fatalf("%s %s", fault, reason)++ }++ f, err = Decode([]byte(`{"v":1,"seq":1,"ts":1,"type":"job","job":{"id":7}}`))++ if err != nil {++ t.Fatal(err)++ }++ if _, reason := f.Msg.(JobRequest).Check(nil); reason != JobBadID {++ t.Fatalf("numeric id: %q", reason)++ }++}nnew file mode 100644@@@ -0,0 +1,9 @@++# OpenVibe.Contracts job-frame fixtures++++Pinned to **OpenVibe.Contracts `4029434`** (PR #7, plan T14 Run): each `.json` file is a verbatim copy of++`fixtures/platform.job-frame/valid/<name>.json`, one frame of `contracts/platform/job-frame.v1.json`++(`platform.job-frame@1`, whose `job` field is `platform.job@1`).++++Used by `internal/protocol` (`TestJobFrameFixtures`): every frame decodes, carries the fields the Node reads, and++encodes back to the same JSON object, so the Node's field names (`chunk_seq`, not `seq`, on `job_stdout`) are the++contract's. When the contract changes, copy the new fixtures here, update the SHA above, and let the test say what moved.nnew file mode 100644@@@ -0,0 +1,7 @@++{++ "v": 1,++ "seq": 8,++ "ts": 1790935201150,++ "type": "job_cancel",++ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ"++}nnew file mode 100644@@@ -0,0 +1,7 @@++{++ "v": 1,++ "seq": 9,++ "ts": 1790935202650,++ "type": "job_exit_ack",++ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ"++}nnew file mode 100644@@@ -0,0 +1,13 @@++{++ "v": 1,++ "seq": 9,++ "ts": 1790935260250,++ "type": "job_exit",++ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ",++ "reason": "ttl",++ "code": null,++ "result": null,++ "usage": {++ "wall_ms": 0++ }++}nnew file mode 100644@@@ -0,0 +1,18 @@++{++ "v": 1,++ "seq": 15,++ "ts": 1790935202600,++ "type": "job_exit",++ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ",++ "reason": "exited",++ "code": 0,++ "result": {++ "url": "media:object-7"++ },++ "usage": {++ "started_ms": 1790935200250,++ "wall_ms": 2350,++ "cpu_ms": 1800,++ "mem_peak_bytes": 41943040++ }++}nnew file mode 100644@@@ -0,0 +1,8 @@++{++ "v": 1,++ "seq": 11,++ "ts": 1790935200255,++ "type": "job_started",++ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ",++ "started_ms": 1790935200250++}nnew file mode 100644@@@ -0,0 +1,9 @@++{++ "v": 1,++ "seq": 13,++ "ts": 1790935201350,++ "type": "job_stdout",++ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ",++ "chunk_seq": 1,++ "chunk": "resized 1 of 1\n"++}nnew file mode 100644@@@ -0,0 +1,10 @@++{++ "v": 1,++ "seq": 12,++ "ts": 1790935201250,++ "type": "job_usage",++ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ",++ "started_ms": 1790935200250,++ "second": 0,++ "cpu_ms": 640++}nnew file mode 100644@@@ -0,0 +1,23 @@++{++ "v": 1,++ "seq": 7,++ "ts": 1790935200000,++ "type": "job",++ "job": {++ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ",++ "class": "function",++ "artifact": {++ "name": "thumbnail",++ "version": "1.2.0"++ },++ "args": {++ "width": 320++ },++ "ttl_ms": 60000,++ "limits": {++ "wall_ms": 30000,++ "cpu_ms": 30000,++ "mem_bytes": 268435456++ }++ }++}
@@ -178,6 +178,41 @@ What the Node does with a command, in order:
@@ -178,6 +178,41 @@ What the Node does with a command, in order:
Commands are never queued while offline and never replayed after a reconnect.
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.
+**Running jobs is not implemented yet:** the Node advertises no runtime class, so it 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_cancel` | `id` |
+| server → device | `job_exit_ack` | `id` |
+| device → server | `job_started` | `id`, `started_ms` |
+| device → server | `job_stdout` | `id`, `chunk_seq` (the job's own counter from 1; the envelope owns `seq`), `chunk` |
+| device → server | `job_usage` | `id`, `started_ms`, `second`, `cpu_ms` |
+| device → server | `job_exit` | `id`, `reason` (`exited`, `cancelled`, `ttl`, `limit`, `stopped`, `failed`), `code`, `result`, `usage` {`started_ms`, `wall_ms`, `cpu_ms`, `mem_peak_bytes`} |
+
+A `job` is answered `ack` (accepted) or `nack` keyed by the job id. A nack's `message` names the first problem found,
+checked in this order:
+
+| `message` | `fault_code` | when |
+|----------------------------------------------|---------------|-------------------------------------------------------------|
+| `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) |
+| `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 (today: every class) |
+
+A resent `job` with an id already answered gets the same answer again, never a second acceptance; the Node
+remembers the last 1024 well-formed ids. `job_cancel` for an id that is not running (unknown, refused or ended) and
+`job_exit_ack` for an unknown id are ignored, with no answer: an `ack` keyed by a job id means the job was accepted.
+A Node that runs jobs lists its classes in `status.capabilities.worker` as `{"runtime_classes": ["function"]}`; the
+key is absent while it runs none, and `worker` is reserved as a plugin name.
+
### Fault codes
### Fault codes
`bad_frame`, `bad_value`, `unsupported`, `not_allowed`, `estopped`, `local_stop`, `expired`, `no_heartbeat`,
`bad_frame`, `bad_value`, `unsupported`, `not_allowed`, `estopped`, `local_stop`, `expired`, `no_heartbeat`,
@@ -212,6 +212,9 @@ func (c *Config) Validate() error {
@@ -212,6 +212,9 @@ func (c *Config) Validate() error {
if p.Name == "" || strings.ContainsAny(p.Name, " /\\") {
if p.Name == "" || strings.ContainsAny(p.Name, " /\\") {
return fmt.Errorf("config: plugin name %q is empty or has spaces or slashes", p.Name)
return fmt.Errorf("config: plugin name %q is empty or has spaces or slashes", p.Name)
}
}
+ if p.Name == "worker" { // status.capabilities.worker holds the job runtime classes
+ return fmt.Errorf("config: plugin name %q is reserved", p.Name)
+ }
if seen[p.Name] {
if seen[p.Name] {
return fmt.Errorf("config: plugin %q listed twice", p.Name)
return fmt.Errorf("config: plugin %q listed twice", p.Name)
}
}
@@ -65,3 +65,12 @@ func TestPluginCommand(t *testing.T) {
@@ -65,3 +65,12 @@ func TestPluginCommand(t *testing.T) {
t.Fatalf("%v", got)
t.Fatalf("%v", got)
}
}
}
}
+
+// TestWorkerNameReserved: status.capabilities.worker holds the job runtime classes, so no plugin may be named worker.
+func TestWorkerNameReserved(t *testing.T) {
+ c := Default()
+ c.Plugins = append(c.Plugins, PluginConfig{Name: "worker", Command: []string{"/bin/true"}})
+ if err := c.Validate(); err == nil {
+ t.Fatal("a plugin named worker was accepted")
+ }
+}
@@ -364,6 +364,15 @@ func (c *Conn) SendError(code, detail string) error {
@@ -364,6 +364,15 @@ func (c *Conn) SendError(code, detail string) error {
return c.Send(protocol.Error{Code: code, Detail: detail})
return c.Send(protocol.Error{Code: code, Detail: detail})
}
}
+// Job sends a job frame; the device answers ack or nack keyed by j.ID (see Reply).
+func (c *Conn) Job(j protocol.Job) error { return c.Send(protocol.JobRequest{Job: j}) }
+
+// JobCancel asks the device to stop job id.
+func (c *Conn) JobCancel(id string) error { return c.Send(protocol.JobCancel{ID: id}) }
+
+// JobExitAck tells the device job id's job_exit is recorded.
+func (c *Conn) JobExitAck(id string) error { return c.Send(protocol.JobExitAck{ID: id}) }
+
// setEstop records the robot's latch: Bot keeps it from the estop frames it sends and the latched:true estop_state
// setEstop records the robot's latch: Bot keeps it from the estop frames it sends and the latched:true estop_state
// frames it receives.
// frames it receives.
func (c *Conn) setEstop(latched bool) {
func (c *Conn) setEstop(latched bool) {
new file mode 100644
new file mode 100644
@@ -0,0 +1,85 @@
@@ -0,0 +1,85 @@
+package node
+
+import (
+ "sync"
+
+ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"
+)
+
+// jobSeenMax bounds how many job ids the Node remembers answers for.
+const jobSeenMax = 1024
+
+// jobs answers job frames. Nothing runs yet: a job is refused unless its class is advertised, and an accepted job is
+// only remembered (the worker that runs it fills classes and executes accepted jobs).
+type jobs struct {
+ mu sync.Mutex
+ classes []string // runtime classes this Node runs and advertises; empty: none
+ answers map[string]protocol.Message // job id → the ack or nack it got, resent unchanged for a repeated job
+ order []string // ids in answers, oldest first
+}
+
+func newJobs(classes []string) *jobs {
+ return &jobs{classes: append([]string(nil), classes...), answers: map[string]protocol.Message{}}
+}
+
+func (j *jobs) runtimeClasses() []string {
+ j.mu.Lock()
+ defer j.mu.Unlock()
+ return append([]string(nil), j.classes...)
+}
+
+// request answers a job frame. A job id already answered gets the same answer again, never a second acceptance;
+// a job whose id is malformed is nacked but not remembered, so junk cannot evict real ids.
+func (j *jobs) request(r protocol.JobRequest) (answer protocol.Message, accepted bool) {
+ j.mu.Lock()
+ defer j.mu.Unlock()
+ id := r.Job.ID
+ if a, ok := j.answers[id]; ok {
+ return a, false
+ }
+ fault, reason := r.Check(j.classes)
+ if fault != "" {
+ answer = protocol.Nack{ID: id, FaultCode: fault, Message: reason}
+ } else {
+ answer, accepted = protocol.Ack{ID: id}, true
+ }
+ if !protocol.ValidJobID(id) {
+ return answer, false
+ }
+ if len(j.order) >= jobSeenMax {
+ delete(j.answers, j.order[0])
+ j.order = j.order[1:]
+ }
+ j.answers[id] = answer
+ j.order = append(j.order, id)
+ return answer, accepted
+}
+
+// accepted reports whether id was accepted (and so would be running).
+func (j *jobs) accepted(id string) bool {
+ j.mu.Lock()
+ defer j.mu.Unlock()
+ _, ok := j.answers[id].(protocol.Ack)
+ return ok
+}
+
+// job answers a job frame with ack or nack, keyed by the job id.
+func (n *Node) job(r protocol.JobRequest) {
+ answer, accepted := n.jobs.request(r)
+ if nk, ok := answer.(protocol.Nack); ok {
+ n.log.Info("job refused", "id", r.Job.ID, "class", r.Job.Class, "reason", nk.Message, "err", r.Err)
+ } else if accepted {
+ n.log.Info("job accepted", "id", r.Job.ID, "class", r.Job.Class)
+ }
+ n.send(answer)
+}
+
+// jobCancel stops an accepted job. An unknown, refused or ended id is ignored, as platform.job-frame@1 says, and no
+// frame answers it: an ack keyed by a job id means the job was accepted.
+func (n *Node) jobCancel(c protocol.JobCancel) {
+ if !n.jobs.accepted(c.ID) {
+ n.log.Debug("job_cancel for a job that is not running ignored", "id", c.ID)
+ return
+ }
+ n.log.Warn("job_cancel: job execution is not implemented, nothing to stop", "id", c.ID)
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,171 @@
@@ -0,0 +1,171 @@
+//go:build !windows
+
+package node
+
+import (
+ "encoding/json"
+ "fmt"
+ "testing"
+ "time"
+
+ "github.com/OpenVibers/OpenVibe.Node/internal/protocol"
+)
+
+func jobID(n int) string { return fmt.Sprintf("job_01JAB2C3D4E5F6G7H8J9K%05d", n) }
+
+func testJob(id string) protocol.Job {
+ return protocol.Job{ID: id, Class: protocol.ClassFunction, Artifact: &protocol.Artifact{Name: "thumbnail", Version: "1.2.0"},
+ Args: json.RawMessage(`{"width":320}`), TTLMS: 60000, Limits: protocol.JobLimits{WallMS: 30000, CPUMS: 30000, MemBytes: 268435456}}
+}
+
+func (e *env) jobReply(j protocol.Job) protocol.Message {
+ e.t.Helper()
+ if err := e.conn.Job(j); err != nil {
+ e.t.Fatal(err)
+ }
+ r, err := e.conn.Reply(j.ID, 5*time.Second)
+ if err != nil {
+ e.t.Fatal(err)
+ }
+ return r
+}
+
+func (e *env) jobNack(j protocol.Job, reason string) protocol.Nack {
+ e.t.Helper()
+ r := e.jobReply(j)
+ nk, ok := r.(protocol.Nack)
+ if !ok || nk.ID != j.ID || nk.Message != reason || nk.FaultCode == "" {
+ e.t.Fatalf("job %s: %+v, want a nack %q", j.ID, r, reason)
+ }
+ return nk
+}
+
+// TestJobsRefused: with no runtime class advertised (nothing executes yet) every job is nacked, with the first problem
+// as the reason, a resent job id gets the same nack, and job_cancel, job_exit_ack and unknown frames change nothing.
+func TestJobsRefused(t *testing.T) {
+ e := start(t)
+ for _, f := range e.conn.Frames() {
+ if s, ok := f.Msg.(protocol.Status); ok {
+ if _, ok := s.Capabilities[protocol.CapWorker]; ok {
+ t.Fatalf("worker capability advertised with no runtime class: %+v", s.Capabilities)
+ }
+ }
+ }
+
+ first := e.jobNack(testJob(jobID(1)), protocol.JobNotAvailable)
+ if first.FaultCode != protocol.FaultUnsupported {
+ t.Fatalf("%+v", first)
+ }
+ if again := e.jobNack(testJob(jobID(1)), protocol.JobNotAvailable); again != first {
+ t.Fatalf("resent job answered %+v, first %+v", again, first)
+ }
+
+ j := testJob(jobID(2))
+ j.Class = "quantum"
+ e.jobNack(j, protocol.JobUnknownClass)
+
+ j = testJob(jobID(3))
+ j.Net = "allow"
+ e.jobNack(j, protocol.JobNetUnsupported)
+
+ e.jobNack(testJob("job_123"), protocol.JobBadID)
+
+ j = testJob(jobID(4))
+ j.Artifact.Version = ">=1.0"
+ e.jobNack(j, protocol.JobNoArtifact)
+
+ // A body that does not decode is nacked by its id, not dropped.
+ bad := fmt.Sprintf(`{"v":1,"seq":900,"ts":%d,"type":"job","job":{"id":%q,"class":"function","ttl_ms":"soon"}}`, time.Now().UnixMilli(), jobID(5))
+ if err := e.conn.SendRaw([]byte(bad)); err != nil {
+ t.Fatal(err)
+ }
+ if r, err := e.conn.Reply(jobID(5), 5*time.Second); err != nil || r.(protocol.Nack).Message != protocol.JobBadFrame {
+ t.Fatalf("%+v %v", r, err)
+ }
+
+ // An unknown frame type, job_cancel and job_exit_ack for unknown ids: ignored, nothing answers them, and the next job
+ // is still answered.
+ if err := e.conn.SendRaw([]byte(fmt.Sprintf(`{"v":1,"seq":901,"ts":%d,"type":"future","id":%q}`, time.Now().UnixMilli(), jobID(6)))); err != nil {
+ t.Fatal(err)
+ }
+ if err := e.conn.JobCancel(jobID(6)); err != nil {
+ t.Fatal(err)
+ }
+ if err := e.conn.JobCancel(jobID(1)); err != nil { // refused earlier: not running either
+ t.Fatal(err)
+ }
+ if err := e.conn.JobExitAck(jobID(6)); err != nil {
+ t.Fatal(err)
+ }
+ e.jobNack(testJob(jobID(7)), protocol.JobNotAvailable)
+ nacks1 := 0
+ for _, f := range e.conn.Frames() {
+ switch m := f.Msg.(type) {
+ case protocol.Ack:
+ t.Fatalf("ack with nothing accepted: %+v", m)
+ case protocol.Nack:
+ if m.ID == jobID(6) {
+ t.Fatalf("job_cancel or job_exit_ack answered: %+v", m)
+ }
+ if m.ID == jobID(1) {
+ nacks1++
+ }
+ }
+ }
+ if nacks1 != 2 {
+ t.Fatalf("job %s answered %d times, want 2 (one per job frame)", jobID(1), nacks1)
+ }
+}
+
+// TestJobAcceptedOnce: once a class is advertised a job is acked, a resent one is acked again (never a second
+// acceptance), and status.capabilities.worker lists the classes.
+func TestJobAcceptedOnce(t *testing.T) {
+ e := start(t)
+ e.node.jobs.mu.Lock()
+ e.node.jobs.classes = []string{protocol.ClassFunction}
+ e.node.jobs.mu.Unlock()
+
+ if r := e.jobReply(testJob(jobID(1))); r != (protocol.Ack{ID: jobID(1)}) {
+ t.Fatalf("%+v", r)
+ }
+ if _, accepted := e.node.jobs.request(protocol.JobRequest{Job: testJob(jobID(1))}); accepted {
+ t.Fatal("a resent job was accepted twice")
+ }
+ if r := e.jobReply(testJob(jobID(1))); r != (protocol.Ack{ID: jobID(1)}) {
+ t.Fatalf("resent: %+v", r)
+ }
+ if err := e.conn.JobCancel(jobID(1)); err != nil {
+ t.Fatal(err)
+ }
+
+ // The next connection's status advertises the class.
+ e.conn.Close()
+ e.conn = e.nextConn()
+ _, err := e.conn.Expect(protocol.TypeStatus, 10*time.Second, func(m protocol.Message) bool {
+ w, _ := m.(protocol.Status).Capabilities[protocol.CapWorker].(map[string]any)
+ c, _ := w["runtime_classes"].([]any)
+ return len(c) == 1 && c[0] == protocol.ClassFunction
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+}
+
+// TestJobAnswersBounded: the Node remembers the answers of the last jobSeenMax valid job ids only, and never a
+// malformed one.
+func TestJobAnswersBounded(t *testing.T) {
+ j := newJobs(nil)
+ for i := 0; i <= jobSeenMax; i++ {
+ j.request(protocol.JobRequest{Job: testJob(jobID(i))})
+ }
+ j.request(protocol.JobRequest{Job: testJob("job_bad")})
+ if len(j.answers) != jobSeenMax || len(j.order) != jobSeenMax {
+ t.Fatalf("%d answers, %d ids", len(j.answers), len(j.order))
+ }
+ if _, ok := j.answers[jobID(0)]; ok {
+ t.Fatal("the oldest id was not forgotten")
+ }
+ if _, ok := j.answers[jobID(jobSeenMax)]; !ok {
+ t.Fatal("the newest id was forgotten")
+ }
+}
@@ -42,6 +42,10 @@ type Options struct {
@@ -42,6 +42,10 @@ type Options struct {
Log *slog.Logger
Log *slog.Logger
Version string
Version string
+ // RuntimeClasses are the job classes this Node runs and advertises in status.capabilities.worker. None yet:
+ // every job is refused "class not available" until the worker that executes jobs fills this in.
+ RuntimeClasses []string
+
// For tests.
// For tests.
HeartbeatInterval time.Duration
HeartbeatInterval time.Duration
LinkBackoffMin time.Duration
LinkBackoffMin time.Duration
@@ -55,6 +59,7 @@ type Node struct {
@@ -55,6 +59,7 @@ type Node struct {
log *slog.Logger
log *slog.Logger
latch *safety.Latch
latch *safety.Latch
dedup *safety.Dedup
dedup *safety.Dedup
+ jobs *jobs
mgr *plugins.Manager
mgr *plugins.Manager
link *link.Link
link *link.Link
pub *video.Publisher
pub *video.Publisher
@@ -93,7 +98,7 @@ func New(opt Options) (*Node, error) {
@@ -93,7 +98,7 @@ func New(opt Options) (*Node, error) {
opt.Log.Error("latch", "err", err)
opt.Log.Error("latch", "err", err)
}
}
n := &Node{opt: opt, log: opt.Log, latch: latch, dedup: safety.NewDedup(4096, 10*time.Minute),
n := &Node{opt: opt, log: opt.Log, latch: latch, dedup: safety.NewDedup(4096, 10*time.Minute),
- telemetry: map[string]map[string]any{}, telemMS: 500}
+ jobs: newJobs(opt.RuntimeClasses), telemetry: map[string]map[string]any{}, telemMS: 500}
n.limits = safety.Merge(protocol.Limits{}, opt.Config.Limits.MaxSpeed, opt.Config.Limits.MaxTurn, opt.Config.Limits.MaxCommandMS)
n.limits = safety.Merge(protocol.Limits{}, opt.Config.Limits.MaxSpeed, opt.Config.Limits.MaxTurn, opt.Config.Limits.MaxCommandMS)
var specs []plugins.Spec
var specs []plugins.Spec
@@ -226,6 +231,13 @@ func (n *Node) Frame(f protocol.Frame) {
@@ -226,6 +231,13 @@ func (n *Node) Frame(f protocol.Frame) {
n.remoteClear(m.By)
n.remoteClear(m.By)
}
}
n.sendEstopState()
n.sendEstopState()
+ case protocol.JobRequest:
+ n.job(m)
+ case protocol.JobCancel:
+ n.jobCancel(m)
+ case protocol.JobExitAck:
+ // No job runs yet, so no job_exit is ever waiting for this.
+ n.log.Debug("job_exit_ack ignored", "id", m.ID)
}
}
}
}
@@ -772,6 +784,9 @@ func (n *Node) sendStatus() {
@@ -772,6 +784,9 @@ func (n *Node) sendStatus() {
s.Faults = append(s.Faults, *i.Fault)
s.Faults = append(s.Faults, *i.Fault)
}
}
}
}
+ if c := n.jobs.runtimeClasses(); len(c) > 0 {
+ s.Capabilities[protocol.CapWorker] = protocol.WorkerCapabilities{RuntimeClasses: c}
+ }
if n.pub != nil {
if n.pub != nil {
s.Video = n.pub.State()
s.Video = n.pub.State()
}
}
@@ -11,6 +11,7 @@ import (
@@ -11,6 +11,7 @@ import (
"encoding/json"
"encoding/json"
"errors"
"errors"
"fmt"
"fmt"
+ "regexp"
)
)
// Version is the envelope version, the `v` field of every frame.
// Version is the envelope version, the `v` field of every frame.
@@ -55,6 +56,59 @@ const (
@@ -55,6 +56,59 @@ const (
TypeEstopState = "estop_state"
TypeEstopState = "estop_state"
)
)
+// Job frames (OpenVibe.Contracts platform.job-frame@1). Server → device: job, job_cancel, job_exit_ack. Device →
+// server: job_started, job_stdout, job_usage, job_exit. A job is accepted or refused with ack/nack keyed by its id.
+const (
+ TypeJob = "job"
+ TypeJobCancel = "job_cancel"
+ TypeJobExitAck = "job_exit_ack"
+ TypeJobStarted = "job_started"
+ TypeJobStdout = "job_stdout"
+ TypeJobUsage = "job_usage"
+ TypeJobExit = "job_exit"
+)
+
+// Runtime classes (platform.runtime-class@1).
+const (
+ ClassFunction = "function"
+ ClassCode = "code"
+ ClassBrowser = "browser"
+ ClassLinux = "linux"
+ ClassDesktop = "desktop"
+ ClassGPU = "gpu"
+)
+
+// RuntimeClasses lists every class the contract names; a worker runs only those it advertises.
+var RuntimeClasses = []string{ClassFunction, ClassCode, ClassBrowser, ClassLinux, ClassDesktop, ClassGPU}
+
+// NetDeny is the only job network policy (and the default when net is absent).
+const NetDeny = "deny"
+
+// CapWorker is the status.capabilities key a worker advertises its runtime classes under (WorkerCapabilities). It is
+// present only when the Node runs at least one class.
+const CapWorker = "worker"
+
+// Job refusal reasons, carried in nack.message (nack.fault_code is bad_value, bad_frame or unsupported).
+const (
+ JobBadID = "missing or malformed job id"
+ JobBadFrame = "malformed job"
+ JobUnknownClass = "unknown class"
+ JobNoArtifact = "function 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"
+)
+
+// job_exit reasons.
+const (
+ ExitExited = "exited"
+ ExitCancelled = "cancelled"
+ ExitTTL = "ttl"
+ ExitLimit = "limit"
+ ExitStopped = "stopped"
+ ExitFailed = "failed"
+)
+
// Command kinds.
// Command kinds.
const (
const (
KindDrive = "drive"
KindDrive = "drive"
@@ -185,8 +239,160 @@ type Error struct {
@@ -185,8 +239,160 @@ type Error struct {
Detail string `json:"detail,omitempty"`
Detail string `json:"detail,omitempty"`
}
}
+// JobLimits are the caps the worker enforces on a running job.
+type JobLimits struct {
+ WallMS int64 `json:"wall_ms"`
+ CPUMS int64 `json:"cpu_ms"`
+ MemBytes int64 `json:"mem_bytes"`
+}
+
+// Artifact is the pre-registered artifact a function job runs, at one exact version.
+type Artifact struct {
+ Name string `json:"name"`
+ Version string `json:"version"`
+}
+
+// Job is platform.job@1.
+type Job struct {
+ ID string `json:"id"` // job_<ULID>, minted by the dispatcher: the idempotency and ack key
+ Class string `json:"class"`
+ Artifact *Artifact `json:"artifact,omitempty"`
+ Args json.RawMessage `json:"args"`
+ TTLMS int64 `json:"ttl_ms"`
+ Limits JobLimits `json:"limits"`
+ Net string `json:"net,omitempty"` // absent means deny
+}
+
+// JobRequest is the `job` frame: run Job. A body that does not decode still yields the frame, with Err set and the
+// id when it is a string, so the Node can nack it rather than drop it (the server resends a job until it is answered).
+type JobRequest struct {
+ Job Job `json:"job"`
+ Err error `json:"-"`
+}
+
+func (r *JobRequest) UnmarshalJSON(b []byte) error {
+ var outer struct {
+ Job json.RawMessage `json:"job"`
+ }
+ if err := json.Unmarshal(b, &outer); err != nil {
+ return err
+ }
+ *r = JobRequest{}
+ if len(outer.Job) == 0 {
+ return nil
+ }
+ if err := json.Unmarshal(outer.Job, &r.Job); err != nil {
+ var id struct {
+ ID any `json:"id"`
+ }
+ _ = json.Unmarshal(outer.Job, &id)
+ s, _ := id.ID.(string)
+ r.Job, r.Err = Job{ID: s}, err
+ }
+ return nil
+}
+
+var (
+ jobIDRe = regexp.MustCompile(`^job_[0-9A-HJKMNP-TV-Z]{26}$`)
+ artifactRe = regexp.MustCompile(`^[a-z0-9][a-z0-9._-]{0,127}$`)
+ artifactVerRe = regexp.MustCompile(`^[0-9A-Za-z][0-9A-Za-z.+-]{0,63}$`)
+)
+
+// ValidJobID reports whether id is a job_<ULID>.
+func ValidJobID(id string) bool { return jobIDRe.MatchString(id) }
+
+// Check returns the fault code and reason to nack this job with, or two empty strings when a worker running the
+// classes in available takes it. The checks run in this order, so the reason names the first problem.
+func (r JobRequest) Check(available []string) (fault, reason string) {
+ j := r.Job
+ switch {
+ case !ValidJobID(j.ID):
+ return FaultBadValue, JobBadID
+ case r.Err != nil:
+ 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)):
+ return FaultBadValue, JobNoArtifact
+ case j.Net != "" && j.Net != NetDeny:
+ return FaultUnsupported, JobNetUnsupported
+ case !isObject(j.Args) || j.TTLMS < 1 || j.Limits.WallMS < 1 || j.Limits.CPUMS < 1 || j.Limits.MemBytes < 1:
+ return FaultBadValue, JobBadLimits
+ case !contains(available, j.Class):
+ return FaultUnsupported, JobNotAvailable
+ }
+ return "", ""
+}
+
+func contains(list []string, s string) bool {
+ for _, v := range list {
+ if v == s {
+ return true
+ }
+ }
+ return false
+}
+
+func isObject(b json.RawMessage) bool {
+ b = bytes.TrimSpace(b)
+ return len(b) >= 2 && b[0] == '{'
+}
+
+// JobCancel stops a job; an unknown or ended id is ignored.
+type JobCancel struct {
+ ID string `json:"id"`
+}
+
+// JobExitAck tells the worker every usage reading of the job is recorded, so it stops resending job_exit.
+type JobExitAck struct {
+ ID string `json:"id"`
+}
+
// ---- device → server ----
// ---- device → server ----
+// JobStarted: the job's process started at StartedMS (worker wall clock, Unix ms), which anchors every usage second.
+type JobStarted struct {
+ ID string `json:"id"`
+ StartedMS int64 `json:"started_ms"`
+}
+
+// JobStdout is a chunk of the job's stdout. ChunkSeq is the job's own counter from 1 (the envelope owns seq).
+type JobStdout struct {
+ ID string `json:"id"`
+ ChunkSeq uint64 `json:"chunk_seq"`
+ Chunk string `json:"chunk"`
+}
+
+// JobUsage: wall-clock second Second (from 0 at StartedMS) of the job has fully elapsed. CPUMS is informational.
+type JobUsage struct {
+ ID string `json:"id"`
+ StartedMS int64 `json:"started_ms"`
+ Second int64 `json:"second"`
+ CPUMS *int64 `json:"cpu_ms,omitempty"`
+}
+
+// JobExitUsage is a job_exit's authoritative usage; StartedMS is set whenever WallMS > 0.
+type JobExitUsage struct {
+ StartedMS *int64 `json:"started_ms,omitempty"`
+ WallMS int64 `json:"wall_ms"`
+ CPUMS *int64 `json:"cpu_ms,omitempty"`
+ MemPeakBytes *int64 `json:"mem_peak_bytes,omitempty"`
+}
+
+// JobExit: the job ended. Code is null when the process was killed or never ran; Result is null unless it exited 0.
+type JobExit struct {
+ ID string `json:"id"`
+ Reason string `json:"reason"`
+ Code *int `json:"code"`
+ Result json.RawMessage `json:"result"`
+ Usage JobExitUsage `json:"usage"`
+}
+
+// WorkerCapabilities is status.capabilities[CapWorker]: the runtime classes this Node runs.
+type WorkerCapabilities struct {
+ RuntimeClasses []string `json:"runtime_classes"`
+}
+
type DriverStatus struct {
type DriverStatus struct {
Name string `json:"name"`
Name string `json:"name"`
Driver string `json:"driver,omitempty"`
Driver string `json:"driver,omitempty"`
@@ -271,6 +477,13 @@ func (Ack) MessageType() string { return TypeAck }
@@ -271,6 +477,13 @@ func (Ack) MessageType() string { return TypeAck }
func (Nack) MessageType() string { return TypeNack }
func (Nack) MessageType() string { return TypeNack }
func (Heartbeat) MessageType() string { return TypeHeartbeat }
func (Heartbeat) MessageType() string { return TypeHeartbeat }
func (EstopState) MessageType() string { return TypeEstopState }
func (EstopState) MessageType() string { return TypeEstopState }
+func (JobRequest) MessageType() string { return TypeJob }
+func (JobCancel) MessageType() string { return TypeJobCancel }
+func (JobExitAck) MessageType() string { return TypeJobExitAck }
+func (JobStarted) MessageType() string { return TypeJobStarted }
+func (JobStdout) MessageType() string { return TypeJobStdout }
+func (JobUsage) MessageType() string { return TypeJobUsage }
+func (JobExit) MessageType() string { return TypeJobExit }
var registry = map[string]func() Message{
var registry = map[string]func() Message{
TypeHello: func() Message { return &Hello{} },
TypeHello: func() Message { return &Hello{} },
@@ -285,6 +498,13 @@ var registry = map[string]func() Message{
@@ -285,6 +498,13 @@ var registry = map[string]func() Message{
TypeNack: func() Message { return &Nack{} },
TypeNack: func() Message { return &Nack{} },
TypeHeartbeat: func() Message { return &Heartbeat{} },
TypeHeartbeat: func() Message { return &Heartbeat{} },
TypeEstopState: func() Message { return &EstopState{} },
TypeEstopState: func() Message { return &EstopState{} },
+ TypeJob: func() Message { return &JobRequest{} },
+ TypeJobCancel: func() Message { return &JobCancel{} },
+ TypeJobExitAck: func() Message { return &JobExitAck{} },
+ TypeJobStarted: func() Message { return &JobStarted{} },
+ TypeJobStdout: func() Message { return &JobStdout{} },
+ TypeJobUsage: func() Message { return &JobUsage{} },
+ TypeJobExit: func() Message { return &JobExit{} },
}
}
// ErrUnknownType is returned (wrapped) by Decode for a frame whose type is not in this version's message set.
// ErrUnknownType is returned (wrapped) by Decode for a frame whose type is not in this version's message set.
@@ -363,6 +583,20 @@ func derefMessage(m Message) Message {
@@ -363,6 +583,20 @@ func derefMessage(m Message) Message {
return *v
return *v
case *EstopState:
case *EstopState:
return *v
return *v
+ case *JobRequest:
+ return *v
+ case *JobCancel:
+ return *v
+ case *JobExitAck:
+ return *v
+ case *JobStarted:
+ return *v
+ case *JobStdout:
+ return *v
+ case *JobUsage:
+ return *v
+ case *JobExit:
+ return *v
}
}
return m
return m
}
}
@@ -5,6 +5,7 @@ import (
@@ -5,6 +5,7 @@ import (
"errors"
"errors"
"os"
"os"
"path/filepath"
"path/filepath"
+ "reflect"
"testing"
"testing"
)
)
@@ -179,3 +180,135 @@ func TestDecodeRejects(t *testing.T) {
@@ -179,3 +180,135 @@ func TestDecodeRejects(t *testing.T) {
t.Fatalf("want ErrUnknownType, got %v", err)
t.Fatalf("want ErrUnknownType, got %v", err)
}
}
}
}
+
+// TestJobFrameFixtures decodes every valid platform.job-frame@1 fixture from OpenVibe.Contracts (testdata/contracts),
+// checks the fields the Node reads, and encodes it back to the same JSON object.
+func TestJobFrameFixtures(t *testing.T) {
+ const id = "job_01JAB2C3D4E5F6G7H8J9K0MNPQ"
+ for _, name := range []string{"job", "job-cancel", "job-exit-ack", "job-started", "job-stdout", "job-usage", "job-exit", "job-exit-never-started"} {
+ b, err := os.ReadFile(filepath.Join("testdata", "contracts", name+".json"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ f, err := Decode(b)
+ if err != nil {
+ t.Fatalf("%s: %v", name, err)
+ }
+ switch m := f.Msg.(type) {
+ case JobRequest:
+ j := m.Job
+ if m.Err != nil || j.ID != id || j.Class != ClassFunction || j.Artifact == nil || j.Artifact.Name != "thumbnail" ||
+ j.Artifact.Version != "1.2.0" || j.TTLMS != 60000 || j.Limits != (JobLimits{WallMS: 30000, CPUMS: 30000, MemBytes: 268435456}) || j.Net != "" {
+ t.Errorf("%s: %+v", name, m)
+ }
+ if fault, reason := m.Check([]string{ClassFunction}); fault != "" {
+ t.Errorf("%s: the contract's example is refused: %s %s", name, fault, reason)
+ }
+ case JobCancel:
+ if m.ID != id {
+ t.Errorf("%s: %+v", name, m)
+ }
+ case JobExitAck:
+ if m.ID != id {
+ t.Errorf("%s: %+v", name, m)
+ }
+ case JobStarted:
+ if m.ID != id || m.StartedMS != 1790935200250 {
+ t.Errorf("%s: %+v", name, m)
+ }
+ case JobStdout:
+ if m.ID != id || m.ChunkSeq != 1 || f.Seq != 13 || m.Chunk != "resized 1 of 1\n" {
+ t.Errorf("%s: %+v (envelope seq %d)", name, m, f.Seq)
+ }
+ case JobUsage:
+ if m.ID != id || m.StartedMS != 1790935200250 || m.Second != 0 || m.CPUMS == nil || *m.CPUMS != 640 {
+ t.Errorf("%s: %+v", name, m)
+ }
+ case JobExit:
+ ran := name == "job-exit"
+ if m.ID != id || (ran && (m.Reason != ExitExited || m.Code == nil || *m.Code != 0 || m.Usage.StartedMS == nil || m.Usage.WallMS != 2350)) ||
+ (!ran && (m.Reason != ExitTTL || m.Code != nil || string(m.Result) != "null" || m.Usage.StartedMS != nil || m.Usage.WallMS != 0)) {
+ t.Errorf("%s: %+v", name, m)
+ }
+ default:
+ t.Fatalf("%s: decoded as %T", name, f.Msg)
+ }
+ out, err := Encode(f.Seq, f.TS, f.Msg)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var want, got any
+ _ = json.Unmarshal(b, &want)
+ if err := json.Unmarshal(out, &got); err != nil {
+ t.Fatal(err)
+ }
+ if !reflect.DeepEqual(want, got) {
+ t.Errorf("%s does not round-trip:\n want %v\n got %v", name, want, got)
+ }
+ }
+}
+
+func TestJobCheck(t *testing.T) {
+ const id = "job_01JAB2C3D4E5F6G7H8J9K0MNPQ"
+ valid := func() Job {
+ return Job{ID: id, Class: ClassFunction, Artifact: &Artifact{Name: "thumbnail", Version: "1.2.0"}, Args: json.RawMessage(`{}`),
+ TTLMS: 1000, Limits: JobLimits{WallMS: 1, CPUMS: 1, MemBytes: 1}}
+ }
+ cases := []struct {
+ name string
+ edit func(*Job)
+ available []string
+ reason string
+ }{
+ {"taken", func(*Job) {}, []string{ClassFunction}, ""},
+ {"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},
+ {"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},
+ {"unknown class", func(j *Job) { j.Class = "quantum" }, []string{ClassFunction}, JobUnknownClass},
+ {"no class", func(j *Job) { j.Class = "" }, []string{ClassFunction}, JobUnknownClass},
+ {"no artifact", func(j *Job) { j.Artifact = nil }, []string{ClassFunction}, JobNoArtifact},
+ {"no version", func(j *Job) { j.Artifact.Version = "" }, []string{ClassFunction}, JobNoArtifact},
+ {"version range", func(j *Job) { j.Artifact.Version = "^1.2" }, []string{ClassFunction}, JobNoArtifact},
+ {"version span", func(j *Job) { j.Artifact.Version = "1.2 - 2" }, []string{ClassFunction}, JobNoArtifact},
+ {"bad artifact name", func(j *Job) { j.Artifact.Name = "Thumbnail" }, []string{ClassFunction}, JobNoArtifact},
+ {"net allow", func(j *Job) { j.Net = "allow" }, []string{ClassFunction}, JobNetUnsupported},
+ {"no args", func(j *Job) { j.Args = nil }, []string{ClassFunction}, JobBadLimits},
+ {"args not an object", func(j *Job) { j.Args = json.RawMessage(`[1]`) }, []string{ClassFunction}, JobBadLimits},
+ {"no ttl", func(j *Job) { j.TTLMS = 0 }, []string{ClassFunction}, JobBadLimits},
+ {"no mem limit", func(j *Job) { j.Limits.MemBytes = 0 }, []string{ClassFunction}, JobBadLimits},
+ }
+ for _, c := range cases {
+ j := valid()
+ c.edit(&j)
+ if _, reason := (JobRequest{Job: j}).Check(c.available); reason != c.reason {
+ t.Errorf("%s: reason %q, want %q", c.name, reason, c.reason)
+ }
+ }
+}
+
+// TestJobUndecodableBody: a job body of the wrong shape still decodes as a job frame, with the id and Err, so the
+// Node nacks it instead of dropping it.
+func TestJobUndecodableBody(t *testing.T) {
+ f, err := Decode([]byte(`{"v":1,"seq":1,"ts":1,"type":"job","job":{"id":"job_01JAB2C3D4E5F6G7H8J9K0MNPQ","class":"function","ttl_ms":"soon"}}`))
+ if err != nil {
+ t.Fatal(err)
+ }
+ r := f.Msg.(JobRequest)
+ if r.Err == nil || r.Job.ID != "job_01JAB2C3D4E5F6G7H8J9K0MNPQ" {
+ t.Fatalf("%+v", r)
+ }
+ if fault, reason := r.Check([]string{ClassFunction}); fault != FaultBadFrame || reason != JobBadFrame {
+ t.Fatalf("%s %s", fault, reason)
+ }
+ f, err = Decode([]byte(`{"v":1,"seq":1,"ts":1,"type":"job","job":{"id":7}}`))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, reason := f.Msg.(JobRequest).Check(nil); reason != JobBadID {
+ t.Fatalf("numeric id: %q", reason)
+ }
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,9 @@
@@ -0,0 +1,9 @@
+# OpenVibe.Contracts job-frame fixtures
+
+Pinned to **OpenVibe.Contracts `4029434`** (PR #7, plan T14 Run): each `.json` file is a verbatim copy of
+`fixtures/platform.job-frame/valid/<name>.json`, one frame of `contracts/platform/job-frame.v1.json`
+(`platform.job-frame@1`, whose `job` field is `platform.job@1`).
+
+Used by `internal/protocol` (`TestJobFrameFixtures`): every frame decodes, carries the fields the Node reads, and
+encodes back to the same JSON object, so the Node's field names (`chunk_seq`, not `seq`, on `job_stdout`) are the
+contract's. When the contract changes, copy the new fixtures here, update the SHA above, and let the test say what moved.
new file mode 100644
new file mode 100644
@@ -0,0 +1,7 @@
@@ -0,0 +1,7 @@
+{
+ "v": 1,
+ "seq": 8,
+ "ts": 1790935201150,
+ "type": "job_cancel",
+ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ"
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,7 @@
@@ -0,0 +1,7 @@
+{
+ "v": 1,
+ "seq": 9,
+ "ts": 1790935202650,
+ "type": "job_exit_ack",
+ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ"
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,13 @@
@@ -0,0 +1,13 @@
+{
+ "v": 1,
+ "seq": 9,
+ "ts": 1790935260250,
+ "type": "job_exit",
+ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ",
+ "reason": "ttl",
+ "code": null,
+ "result": null,
+ "usage": {
+ "wall_ms": 0
+ }
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,18 @@
@@ -0,0 +1,18 @@
+{
+ "v": 1,
+ "seq": 15,
+ "ts": 1790935202600,
+ "type": "job_exit",
+ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ",
+ "reason": "exited",
+ "code": 0,
+ "result": {
+ "url": "media:object-7"
+ },
+ "usage": {
+ "started_ms": 1790935200250,
+ "wall_ms": 2350,
+ "cpu_ms": 1800,
+ "mem_peak_bytes": 41943040
+ }
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,8 @@
@@ -0,0 +1,8 @@
+{
+ "v": 1,
+ "seq": 11,
+ "ts": 1790935200255,
+ "type": "job_started",
+ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ",
+ "started_ms": 1790935200250
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,9 @@
@@ -0,0 +1,9 @@
+{
+ "v": 1,
+ "seq": 13,
+ "ts": 1790935201350,
+ "type": "job_stdout",
+ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ",
+ "chunk_seq": 1,
+ "chunk": "resized 1 of 1\n"
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,10 @@
@@ -0,0 +1,10 @@
+{
+ "v": 1,
+ "seq": 12,
+ "ts": 1790935201250,
+ "type": "job_usage",
+ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ",
+ "started_ms": 1790935200250,
+ "second": 0,
+ "cpu_ms": 640
+}
new file mode 100644
new file mode 100644
@@ -0,0 +1,23 @@
@@ -0,0 +1,23 @@
+{
+ "v": 1,
+ "seq": 7,
+ "ts": 1790935200000,
+ "type": "job",
+ "job": {
+ "id": "job_01JAB2C3D4E5F6G7H8J9K0MNPQ",
+ "class": "function",
+ "artifact": {
+ "name": "thumbnail",
+ "version": "1.2.0"
+ },
+ "args": {
+ "width": 320
+ },
+ "ttl_ms": 60000,
+ "limits": {
+ "wall_ms": 30000,
+ "cpu_ms": 30000,
+ "mem_bytes": 268435456
+ }
+ }
+}