openvibers/openvibe.link Commits

Add metering and resend contract tests and spell out the run key (#15)

OP OpenVibers committed 3 Oct 2026 · 60c5529f2c · 3 files, +191 −5

local checks ✗ 0/2 · CI ✓ 9/9 · review SHIP by DeepSeek harness

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

Changes

ddiff --git a/docs/protocol.md b/docs/protocol.md
iindex 07d7b7c..6f58804 100644
---- a/docs/protocol.md
++++ b/docs/protocol.md
@@@ -219,11 +219,19 @@ An accepted job runs once the `ack` is out: `job_started` (`started_ms` anchors
`job_stdout` chunks (at most 16 KiB each as sent and `max_output_bytes` in all, valid UTF-8 with invalid bytes as
U+FFFD, `chunk_seq` from 1; best effort, not resent), one `job_usage` for
every wall-clock second once it has fully elapsed (`second` from 0, `cpu_ms` that second's CPU time), then
--`job_exit`. Its `usage` is authoritative for metering (Bot writes `usage-sample` readings `service:"run"`,
--`operation:"function.invoke"`, unit `s`, idempotency key `run:<job id>:<n>`; none of those fields changes on a
--resend). The Node resends `job_started` for running jobs and `job_exit` for ended ones after every reconnect, and
--`job_exit` again on a `job` or `job_cancel` for that id, until `job_exit_ack`. `job_started` goes out only once the
--job's CPU limit is set; a job that fails before (no namespaces, spawn or CPU limit failed) ends `failed` without it.
++`job_exit`. Second `n` is sent once `started_ms + (n + 1) s` has passed (`second` 0 at one whole second, 1 at two,
++and so on): the seconds are contiguous, with no duplicate and no gap, while the job runs.
++
++`job_exit.usage` is **authoritative for metering**. Bot turns each `job_usage` into one `usage-sample`
++(`service:"run"`, `operation:"function.invoke"`, unit `s`) whose idempotency key is `run:<job id>:<n>`, where `<n>`
++is the 0-based wall second — the `second` of the `job_usage` it covers, both counted from `started_ms`. A resent
++`job` or `job_cancel` repeats the same `job_exit` and so the same keys, never a second reading and never a `:<n>`
++for a second already metered. `job_exit.usage.wall_ms` is at least `(n + 1) * 1000` for every second `n` already
++sent, so the exit's own usage covers everything the per-second frames announced — it is never corrected downward
++by the resent `job_exit`. The Node resends `job_started` for running jobs and `job_exit` for ended ones after every
++reconnect, and `job_exit` again on a `job` or `job_cancel` for that id, until `job_exit_ack`. `job_started` goes out
++only once the job's CPU limit is set; a job that fails before (no namespaces, spawn or CPU limit failed) ends
++`failed` without it.
A Node shutting down sends `job_exit` (`stopped`) also for a job acked but not started yet. `job_exit.reason`:
| `reason` | `code` | when |
ddiff --git a/internal/node/worker_test.go b/internal/node/worker_test.go
iindex 8111181..337a635 100644
---- a/internal/node/worker_test.go
++++ b/internal/node/worker_test.go
@@@ -296,3 +296,103 @@ func TestLocalStopEndsJobs(t *testing.T) {
t.Fatalf("status jobs after the end: %+v", jobs)
}
}
++
++// TestResendAndCancelShareOneUsageSet: a job that runs for whole wall-clock seconds sends one job_usage per elapsed
++// second (0 then 1, no gap and no duplicate); a resent job and a job_cancel after it ended both answer with the same
++// job_exit, never a second job_started, a repeated usage second, or a second run of the function.
++func TestResendAndCancelShareOneUsageSet(t *testing.T) {
++ e := startWorkerNode(t)
++ marker := filepath.Join(e.paths.StateDir, "starts")
++ j := testJob(jobID(1))
++ j.Artifact = &protocol.Artifact{Name: "hello", Version: "1.0.0"}
++ j.Args = json.RawMessage(fmt.Sprintf(`{"marker":%q,"hold_ms":2500}`, marker))
++ j.Limits.MemBytes = 4 << 30
++ isExit := func(m protocol.Message) bool {
++ ex, ok := m.(protocol.JobExit)
++ return ok && ex.ID == j.ID
++ }
++
++ 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, isExit)
++ if err != nil {
++ t.Fatal(err)
++ }
++ ex := f.Msg.(protocol.JobExit)
++ if ex.Reason != protocol.ExitExited || ex.Code == nil || *ex.Code != 0 || ex.Usage.StartedMS == nil {
++ t.Fatalf("%+v", ex)
++ }
++
++ // usageSeconds is the `second` of every job_usage for this job, in arrival order, checking each names the same
++ // started_ms as its job_exit and carries a cpu_ms.
++ usageSeconds := func() []int64 {
++ var seconds []int64
++ for _, fr := range e.conn.Frames() {
++ u, ok := fr.Msg.(protocol.JobUsage)
++ if !ok || u.ID != j.ID {
++ continue
++ }
++ if u.StartedMS != *ex.Usage.StartedMS || u.CPUMS == nil {
++ t.Fatalf("usage %+v", u)
++ }
++ seconds = append(seconds, u.Second)
++ }
++ return seconds
++ }
++ jobStarts := func() int {
++ n := 0
++ for _, fr := range e.conn.Frames() {
++ if s, ok := fr.Msg.(protocol.JobStarted); ok && s.ID == j.ID {
++ n++
++ }
++ }
++ return n
++ }
++
++ first := usageSeconds()
++ if len(first) < 2 {
++ t.Fatalf("usage seconds %v, want 0..n-1 for a job that held 2.5 s", first)
++ }
++ for n, second := range first {
++ if second != int64(n) || ex.Usage.WallMS < (second+1)*1000 {
++ t.Fatalf("usage seconds %v, wall_ms %d, want 0..%d contiguous and covering each second sent", first, ex.Usage.WallMS, len(first)-1)
++ }
++ }
++ if n := jobStarts(); n != 1 {
++ t.Fatalf("job_started sent %d times, want 1", n)
++ }
++
++ // A resent job and a job_cancel of the ended job each answer with the same job_exit (same result and usage).
++ for _, resend := range []struct {
++ name string
++ send func() error
++ }{
++ {"resent job", func() error { return e.conn.Job(j) }},
++ {"job_cancel", func() error { return e.conn.JobCancel(j.ID) }},
++ } {
++ if err := resend.send(); err != nil {
++ t.Fatal(err)
++ }
++ f, err := e.conn.Expect(protocol.TypeJobExit, 5*time.Second, isExit)
++ if err != nil {
++ t.Fatalf("%s: %v", resend.name, err)
++ }
++ again := f.Msg.(protocol.JobExit)
++ if again.Reason != ex.Reason || !reflect.DeepEqual(again.Code, ex.Code) || !reflect.DeepEqual(again.Usage, ex.Usage) || string(again.Result) != string(ex.Result) {
++ t.Fatalf("%s answered %+v, first job_exit %+v", resend.name, again, ex)
++ }
++ }
++
++ // No second process, job_started or usage second, and one usage set only.
++ if n := jobStarts(); n != 1 {
++ t.Fatalf("job_started sent %d times after the resends, want 1", n)
++ }
++ if again := usageSeconds(); !slices.Equal(again, first) {
++ t.Fatalf("usage seconds %v after the resends, first %v", again, first)
++ }
++ b, err := os.ReadFile(marker)
++ if err != nil || string(b) != "started\n" {
++ t.Fatalf("the function ran %q (%v), want once", b, err)
++ }
++}
ddiff --git a/internal/worker/worker_test.go b/internal/worker/worker_test.go
iindex 31c8112..4be1818 100644
---- a/internal/worker/worker_test.go
++++ b/internal/worker/worker_test.go
@@@ -8,6 +8,8 @@ import (
"log/slog"
"os"
"os/exec"
++ "runtime"
++ "slices"
"sync"
"testing"
"time"
@@@ -219,3 +221,79 @@ func TestCloseEndsUnlaunched(t *testing.T) {
t.Fatalf("admitted after Close: %q", f)
}
}
++
++// TestUsageHelperProcess is the job's process for TestUsageSecondsContiguous when this test binary runs as the `meter`
++// function (`-- meter`): it holds for hold_ms from its args and exits 0 without writing a result.
++func TestUsageHelperProcess(t *testing.T) {
++ i := slices.Index(os.Args, "--")
++ if i < 0 || i+1 >= len(os.Args) || os.Args[i+1] != "meter" {
++ return
++ }
++ var args struct {
++ HoldMS int64 `json:"hold_ms"`
++ }
++ _ = json.NewDecoder(os.Stdin).Decode(&args)
++ time.Sleep(time.Duration(args.HoldMS) * time.Millisecond)
++ os.Exit(0)
++}
++
++// TestUsageSecondsContiguous: a function that holds for several whole wall-clock seconds after job_started sends
++// exactly one job_usage for each fully elapsed second, `second` 0 then 1 then ... with no gap and no duplicate, and
++// its job_exit's usage is never smaller than the seconds already sent (wall_ms >= (n+1)*1000 for every second n).
++func TestUsageSecondsContiguous(t *testing.T) {
++ if runtime.GOOS != "linux" {
++ t.Skip("function jobs run on Linux only")
++ }
++ exe, err := os.Executable()
++ if err != nil {
++ t.Fatal(err)
++ }
++ // Isolation is not what this test measures: run the helper directly, without the job namespaces.
++ s := &sink{}
++ w := New(config.WorkerConfig{Enabled: true, AllowSameUser: true,
++ Functions: []config.FunctionConfig{{Name: "meter", Version: "1.0.0",
++ Command: []string{exe, "-test.run=^TestUsageHelperProcess$", "--", "meter"}}}}, s.send, quiet())
++ w.isolate = func(*exec.Cmd, config.WorkerConfig) error { return nil }
++ t.Cleanup(w.Close)
++
++ const holdMS = 2500 // two whole seconds, so seconds 0 and 1 are sent before the exit
++ j := testJob(1, "meter")
++ j.Args = json.RawMessage(fmt.Sprintf(`{"hold_ms":%d}`, holdMS))
++ if f, r := w.Admit(j); f != "" {
++ t.Fatalf("%s %s", f, r)
++ }
++ w.Launch(j.ID)
++ ex := s.exit(t, j.ID)
++ if ex.Reason != protocol.ExitExited || ex.Code == nil || *ex.Code != 0 || ex.Usage.StartedMS == nil {
++ t.Fatalf("%+v", ex)
++ }
++ if n := s.count(protocol.TypeJobStarted, j.ID); n != 1 {
++ t.Fatalf("job_started sent %d times, want 1", n)
++ }
++ started := s.wait(t, protocol.TypeJobStarted, j.ID, 1, 5*time.Second).(protocol.JobStarted)
++ if started.StartedMS != *ex.Usage.StartedMS {
++ t.Fatalf("job_started at %d, job_exit usage at %d", started.StartedMS, *ex.Usage.StartedMS)
++ }
++ var seconds []int64
++ for _, m := range s.all() {
++ u, ok := m.(protocol.JobUsage)
++ if !ok || u.ID != j.ID {
++ continue
++ }
++ if u.StartedMS != *ex.Usage.StartedMS || u.CPUMS == nil {
++ t.Fatalf("usage %+v", u)
++ }
++ seconds = append(seconds, u.Second)
++ }
++ if len(seconds) < 2 {
++ t.Fatalf("usage seconds %v, want 0..n-1 for a job that held %d ms", seconds, holdMS)
++ }
++ for n, second := range seconds {
++ if second != int64(n) {
++ t.Fatalf("usage seconds %v, want 0..%d with no gap or duplicate", seconds, len(seconds)-1)
++ }
++ if ex.Usage.WallMS < (second+1)*1000 {
++ t.Fatalf("usage second %d sent but wall_ms is %d, want at least %d", second, ex.Usage.WallMS, (second+1)*1000)
++ }
++ }
++}