detached 924eaca

server-computed runner online state and running version (#40)

commit
924eaca15f3959f330e82bcb8c43d7be4ee0e1ba
author
brice · · 2026-10-16 10:49 UTC
committer
bnema · · 2026-10-16 10:49 UTC
parent
81486c6
signature
SSH UnverifiedSHA256:VOCA5lYARHcbKDZ7GhTV45VmcjWnJ6ZQIBazj17LivA The key is unknown
refs
tag: v0.16.0
13 files · +232 -7
Split
UnifiedSplit
M internal/adapters/kvstore/ci.go +31 -0
@@ -1038,6 +1038,37 @@ func (c *CI) ListRunners(ctx context.Context) ([]domain.CIRunner, error) {
10381038 return out, nil
10391039 }
10401040
1041+// SetRunnerVersion records the version a runner runs now; an unchanged version
1042+// is not written.
1043+func (c *CI) SetRunnerVersion(ctx context.Context, id, version string) error {
1044+ for range createRunTries {
1045+ item, ok := c.db.Get(ctx, runnerKey(id))
1046+ if !ok || id == "" || strings.Contains(id, "/") {
1047+ return domain.ErrNotFound
1048+ }
1049+ r, err := decodeRunner(item.Value)
1050+ if err != nil {
1051+ return err
1052+ }
1053+ if r.Version == version {
1054+ return nil
1055+ }
1056+ r.Version = version
1057+ var b kv.Batch
1058+ b.RequireVersion(runnerKey(id), item.Version)
1059+ b.Put(runnerKey(id), encodeRunner(r))
1060+ _, err = c.db.Apply(ctx, &b)
1061+ if errors.Is(err, kv.ErrConflict) {
1062+ continue
1063+ }
1064+ if err != nil {
1065+ return fmt.Errorf("set runner %q version: %w", id, err)
1066+ }
1067+ return nil
1068+ }
1069+ return fmt.Errorf("set runner %q version: %w", id, domain.ErrConflict)
1070+}
1071+
10411072 // RemoveRunner removes the runner, its name and its token.
10421073 func (c *CI) RemoveRunner(ctx context.Context, id string) (domain.CIRunner, error) {
10431074 for range createRunTries {
@@ -1038,6 +1038,37 @@ func (c *CI) ListRunners(ctx context.Context) ([]domain.CIRunner, error) {
1038 return out, nil1038 return out, nil
1039 }1039 }
1040 1040
1041+// SetRunnerVersion records the version a runner runs now; an unchanged version
1042+// is not written.
1043+func (c *CI) SetRunnerVersion(ctx context.Context, id, version string) error {
1044+ for range createRunTries {
1045+ item, ok := c.db.Get(ctx, runnerKey(id))
1046+ if !ok || id == "" || strings.Contains(id, "/") {
1047+ return domain.ErrNotFound
1048+ }
1049+ r, err := decodeRunner(item.Value)
1050+ if err != nil {
1051+ return err
1052+ }
1053+ if r.Version == version {
1054+ return nil
1055+ }
1056+ r.Version = version
1057+ var b kv.Batch
1058+ b.RequireVersion(runnerKey(id), item.Version)
1059+ b.Put(runnerKey(id), encodeRunner(r))
1060+ _, err = c.db.Apply(ctx, &b)
1061+ if errors.Is(err, kv.ErrConflict) {
1062+ continue
1063+ }
1064+ if err != nil {
1065+ return fmt.Errorf("set runner %q version: %w", id, err)
1066+ }
1067+ return nil
1068+ }
1069+ return fmt.Errorf("set runner %q version: %w", id, domain.ErrConflict)
1070+}
1071+
1041 // RemoveRunner removes the runner, its name and its token.1072 // RemoveRunner removes the runner, its name and its token.
1042 func (c *CI) RemoveRunner(ctx context.Context, id string) (domain.CIRunner, error) {1073 func (c *CI) RemoveRunner(ctx context.Context, id string) (domain.CIRunner, error) {
1043 for range createRunTries {1074 for range createRunTries {
M internal/adapters/kvstore/ci_test.go +24 -0
@@ -251,6 +251,30 @@ func TestCIRunnersCreateReplaceFindRemove(t *testing.T) {
251251 assert.NoError(t, err)
252252 }
253253
254+func TestCISetRunnerVersion(t *testing.T) {
255+ ctx := context.Background()
256+ _, ci, _ := ciRepo(t)
257+ h := domain.HashTokenSecret("ermiterun_v")
258+ _, err := ci.CreateRunner(ctx, domain.CIRunner{ID: "r1", Name: "alpha", Version: "v1", TokenHash: h, RegisteredAt: created}, false)
259+ require.NoError(t, err)
260+
261+ require.NoError(t, ci.SetRunnerVersion(ctx, "r1", "v2"))
262+ got, err := ci.GetRunner(ctx, "r1")
263+ require.NoError(t, err)
264+ assert.Equal(t, "v2", got.Version)
265+ assert.Equal(t, h, got.TokenHash, "the rest of the record is kept")
266+ assert.Equal(t, "alpha", got.Name)
267+
268+ // An unchanged version is not written.
269+ before, ok := ci.RawVersionForTest(ctx, "r1")
270+ require.True(t, ok)
271+ require.NoError(t, ci.SetRunnerVersion(ctx, "r1", "v2"))
272+ after, _ := ci.RawVersionForTest(ctx, "r1")
273+ assert.Equal(t, before, after)
274+
275+ assert.ErrorIs(t, ci.SetRunnerVersion(ctx, "gone", "v1"), domain.ErrNotFound)
276+}
277+
254278 func TestCISweepRemovesOldCompletedRunsOnly(t *testing.T) {
255279 ctx := context.Background()
256280 _, ci, n := ciRepo(t)
@@ -251,6 +251,30 @@ func TestCIRunnersCreateReplaceFindRemove(t *testing.T) {
251 assert.NoError(t, err)251 assert.NoError(t, err)
252 }252 }
253 253
254+func TestCISetRunnerVersion(t *testing.T) {
255+ ctx := context.Background()
256+ _, ci, _ := ciRepo(t)
257+ h := domain.HashTokenSecret("ermiterun_v")
258+ _, err := ci.CreateRunner(ctx, domain.CIRunner{ID: "r1", Name: "alpha", Version: "v1", TokenHash: h, RegisteredAt: created}, false)
259+ require.NoError(t, err)
260+
261+ require.NoError(t, ci.SetRunnerVersion(ctx, "r1", "v2"))
262+ got, err := ci.GetRunner(ctx, "r1")
263+ require.NoError(t, err)
264+ assert.Equal(t, "v2", got.Version)
265+ assert.Equal(t, h, got.TokenHash, "the rest of the record is kept")
266+ assert.Equal(t, "alpha", got.Name)
267+
268+ // An unchanged version is not written.
269+ before, ok := ci.RawVersionForTest(ctx, "r1")
270+ require.True(t, ok)
271+ require.NoError(t, ci.SetRunnerVersion(ctx, "r1", "v2"))
272+ after, _ := ci.RawVersionForTest(ctx, "r1")
273+ assert.Equal(t, before, after)
274+
275+ assert.ErrorIs(t, ci.SetRunnerVersion(ctx, "gone", "v1"), domain.ErrNotFound)
276+}
277+
254 func TestCISweepRemovesOldCompletedRunsOnly(t *testing.T) {278 func TestCISweepRemovesOldCompletedRunsOnly(t *testing.T) {
255 ctx := context.Background()279 ctx := context.Background()
256 _, ci, n := ciRepo(t)280 _, ci, n := ciRepo(t)
M internal/adapters/kvstore/export_test.go +9 -0
@@ -2,6 +2,8 @@
22
33 package kvstore
44
5+import "context"
6+
57 // SetAfterSettingsSnapshot installs the hook that runs between the snapshot of
68 // DeleteEnvironment and its batch, and returns the function that removes it.
79 func SetAfterSettingsSnapshot(f func()) (restore func()) {
@@ -16,3 +18,10 @@ var (
1618 EncodeConcurrencyRecord = encodeConcurrencyRecord
1719 EncodeConcurrencyMark = encodeConcurrencyMark
1820 )
21+
22+// RawVersionForTest is the kv version of a runner record: it moves on every
23+// write.
24+func (c *CI) RawVersionForTest(ctx context.Context, id string) (uint64, bool) {
25+ item, ok := c.db.Get(ctx, runnerKey(id))
26+ return item.Version, ok
27+}
@@ -2,6 +2,8 @@
2 2
3 package kvstore3 package kvstore
4 4
5+import "context"
6+
5 // SetAfterSettingsSnapshot installs the hook that runs between the snapshot of7 // SetAfterSettingsSnapshot installs the hook that runs between the snapshot of
6 // DeleteEnvironment and its batch, and returns the function that removes it.8 // DeleteEnvironment and its batch, and returns the function that removes it.
7 func SetAfterSettingsSnapshot(f func()) (restore func()) {9 func SetAfterSettingsSnapshot(f func()) (restore func()) {
@@ -16,3 +18,10 @@ var (
16 EncodeConcurrencyRecord = encodeConcurrencyRecord18 EncodeConcurrencyRecord = encodeConcurrencyRecord
17 EncodeConcurrencyMark = encodeConcurrencyMark19 EncodeConcurrencyMark = encodeConcurrencyMark
18 )20 )
21+
22+// RawVersionForTest is the kv version of a runner record: it moves on every
23+// write.
24+func (c *CI) RawVersionForTest(ctx context.Context, id string) (uint64, bool) {
25+ item, ok := c.db.Get(ctx, runnerKey(id))
26+ return item.Version, ok
27+}
M internal/adapters/rpc/ci.go +1 -1
@@ -95,7 +95,7 @@ func (s *runnerService) RegisterRunner(ctx context.Context, req *connect.Request
9595 }
9696
9797 func (s *runnerService) FetchJob(ctx context.Context, req *connect.Request[pb.FetchJobRequest]) (*connect.Response[pb.FetchJobResponse], error) {
98- res, err := s.runners.FetchJob(ctx, secretFrom(ctx), ports.FetchJobParams{Labels: req.Msg.GetLabels(), PollID: req.Msg.GetPollId()})
98+ res, err := s.runners.FetchJob(ctx, secretFrom(ctx), ports.FetchJobParams{Labels: req.Msg.GetLabels(), PollID: req.Msg.GetPollId(), Version: req.Msg.GetVersion()})
9999 if err != nil {
100100 return nil, s.failed(ctx, err)
101101 }
@@ -95,7 +95,7 @@ func (s *runnerService) RegisterRunner(ctx context.Context, req *connect.Request
95 }95 }
96 96
97 func (s *runnerService) FetchJob(ctx context.Context, req *connect.Request[pb.FetchJobRequest]) (*connect.Response[pb.FetchJobResponse], error) {97 func (s *runnerService) FetchJob(ctx context.Context, req *connect.Request[pb.FetchJobRequest]) (*connect.Response[pb.FetchJobResponse], error) {
98- res, err := s.runners.FetchJob(ctx, secretFrom(ctx), ports.FetchJobParams{Labels: req.Msg.GetLabels(), PollID: req.Msg.GetPollId()})98+ res, err := s.runners.FetchJob(ctx, secretFrom(ctx), ports.FetchJobParams{Labels: req.Msg.GetLabels(), PollID: req.Msg.GetPollId(), Version: req.Msg.GetVersion()})
99 if err != nil {99 if err != nil {
100 return nil, s.failed(ctx, err)100 return nil, s.failed(ctx, err)
101 }101 }
M internal/adapters/rpc/ci_convert.go +1 -1
@@ -208,7 +208,7 @@ func checkToPB(c domain.CICheck) *pb.Check {
208208 func runnerToPB(r ports.RunnerView) *pb.Runner {
209209 return &pb.Runner{
210210 Id: r.ID, Name: r.Name, Labels: r.Labels, Isolation: isolationToPB(r.Isolation), Version: r.Version,
211- Arch: r.Arch, Os: r.OS, RegisteredAt: timestamp(r.RegisteredAt), LastSeen: timestamp(r.LastSeen), Busy: r.Busy,
211+ Arch: r.Arch, Os: r.OS, RegisteredAt: timestamp(r.RegisteredAt), LastSeen: timestamp(r.LastSeen), Busy: r.Busy, Online: r.Online,
212212 }
213213 }
214214
@@ -208,7 +208,7 @@ func checkToPB(c domain.CICheck) *pb.Check {
208 func runnerToPB(r ports.RunnerView) *pb.Runner {208 func runnerToPB(r ports.RunnerView) *pb.Runner {
209 return &pb.Runner{209 return &pb.Runner{
210 Id: r.ID, Name: r.Name, Labels: r.Labels, Isolation: isolationToPB(r.Isolation), Version: r.Version,210 Id: r.ID, Name: r.Name, Labels: r.Labels, Isolation: isolationToPB(r.Isolation), Version: r.Version,
211- Arch: r.Arch, Os: r.OS, RegisteredAt: timestamp(r.RegisteredAt), LastSeen: timestamp(r.LastSeen), Busy: r.Busy,211+ Arch: r.Arch, Os: r.OS, RegisteredAt: timestamp(r.RegisteredAt), LastSeen: timestamp(r.LastSeen), Busy: r.Busy, Online: r.Online,
212 }212 }
213 }213 }
214 214
M internal/app/ci_flow_test.go +32 -0
@@ -682,3 +682,35 @@ func TestCIGithubContextOfAPush(t *testing.T) {
682682 assert.Equal(t, publicOrigin+"/owner/proj/compare/"+first[:12]+"..."+second[:12], event["compare"])
683683 assert.Equal(t, false, event["created"])
684684 }
685+
686+// ListRunners says by the server's clock whether a runner is online, and shows
687+// the version the runner reported at its last poll, not the one it registered
688+// with. A poll without a version keeps the stored one.
689+func TestCIRunnerListOnlineAndVersion(t *testing.T) {
690+ c := newCIHarness(t, func(cfg *usecase.CIConfig) { cfg.PollTimeout = 50 * time.Millisecond })
691+ r := c.registerRunner("alpha", "ubuntu-latest")
692+ list := func() *pb.Runner {
693+ res, err := c.root.run.ListRunners(c.ctx, connect.NewRequest(&pb.ListRunnersRequest{}))
694+ require.NoError(t, err)
695+ require.Len(t, res.Msg.GetRunners(), 1)
696+ return res.Msg.GetRunners()[0]
697+ }
698+ poll := func(version string) error {
699+ _, err := r.api.runner.FetchJob(c.ctx, connect.NewRequest(&pb.FetchJobRequest{Version: version}))
700+ return err
701+ }
702+ got := list()
703+ assert.False(t, got.GetOnline(), "never seen")
704+ assert.Equal(t, "devel", got.GetVersion())
705+
706+ require.NoError(t, poll("v1.2.3"))
707+ got = list()
708+ assert.True(t, got.GetOnline())
709+ assert.Equal(t, "v1.2.3", got.GetVersion(), "the running version replaces the registered one")
710+
711+ require.NoError(t, poll(""))
712+ assert.Equal(t, "v1.2.3", list().GetVersion(), "an older runner sends none: unchanged")
713+
714+ assert.Equal(t, connect.CodeInvalidArgument, ciErrCode(poll(strings.Repeat("v", 65))))
715+ assert.Equal(t, "v1.2.3", list().GetVersion())
716+}
@@ -682,3 +682,35 @@ func TestCIGithubContextOfAPush(t *testing.T) {
682 assert.Equal(t, publicOrigin+"/owner/proj/compare/"+first[:12]+"..."+second[:12], event["compare"])682 assert.Equal(t, publicOrigin+"/owner/proj/compare/"+first[:12]+"..."+second[:12], event["compare"])
683 assert.Equal(t, false, event["created"])683 assert.Equal(t, false, event["created"])
684 }684 }
685+
686+// ListRunners says by the server's clock whether a runner is online, and shows
687+// the version the runner reported at its last poll, not the one it registered
688+// with. A poll without a version keeps the stored one.
689+func TestCIRunnerListOnlineAndVersion(t *testing.T) {
690+ c := newCIHarness(t, func(cfg *usecase.CIConfig) { cfg.PollTimeout = 50 * time.Millisecond })
691+ r := c.registerRunner("alpha", "ubuntu-latest")
692+ list := func() *pb.Runner {
693+ res, err := c.root.run.ListRunners(c.ctx, connect.NewRequest(&pb.ListRunnersRequest{}))
694+ require.NoError(t, err)
695+ require.Len(t, res.Msg.GetRunners(), 1)
696+ return res.Msg.GetRunners()[0]
697+ }
698+ poll := func(version string) error {
699+ _, err := r.api.runner.FetchJob(c.ctx, connect.NewRequest(&pb.FetchJobRequest{Version: version}))
700+ return err
701+ }
702+ got := list()
703+ assert.False(t, got.GetOnline(), "never seen")
704+ assert.Equal(t, "devel", got.GetVersion())
705+
706+ require.NoError(t, poll("v1.2.3"))
707+ got = list()
708+ assert.True(t, got.GetOnline())
709+ assert.Equal(t, "v1.2.3", got.GetVersion(), "the running version replaces the registered one")
710+
711+ require.NoError(t, poll(""))
712+ assert.Equal(t, "v1.2.3", list().GetVersion(), "an older runner sends none: unchanged")
713+
714+ assert.Equal(t, connect.CodeInvalidArgument, ciErrCode(poll(strings.Repeat("v", 65))))
715+ assert.Equal(t, "v1.2.3", list().GetVersion())
716+}
M internal/app/cli_config_ci_test.go +28 -0
@@ -4,7 +4,9 @@ package app
44
55 import (
66 "context"
7+ "strings"
78 "testing"
9+ "time"
810
911 "connectrpc.com/connect"
1012 "github.com/stretchr/testify/assert"
@@ -35,6 +37,9 @@ func (f *fakeAPI) RerunRun(_ context.Context, r *connect.Request[ermite.RerunRun
3537
3638 func (f *fakeAPI) ListRunners(_ context.Context, r *connect.Request[ermite.ListRunnersRequest]) (*connect.Response[ermite.ListRunnersResponse], error) {
3739 return respond(f, "ListRunners", r, func(*ermite.ListRunnersRequest) (*ermite.ListRunnersResponse, error) {
40+ if f.runners != nil {
41+ return &ermite.ListRunnersResponse{Runners: f.runners}, nil
42+ }
3843 return &ermite.ListRunnersResponse{Runners: []*ermite.Runner{{Id: "r1", Name: "alpha"}}}, nil
3944 })
4045 }
@@ -112,3 +117,26 @@ func TestCICompletionNeverRunsTokenCmd(t *testing.T) {
112117 })
113118 }
114119 }
120+
121+// The CLI shows what the server decided: it never compares last_seen with its
122+// own clock.
123+func TestRunnerListShowsTheServersVerdict(t *testing.T) {
124+ f := &fakeAPI{}
125+ f.runners = []*ermite.Runner{
126+ {Id: "r1", Name: "up", Version: "v2", Online: true},
127+ {Id: "r2", Name: "busy", Online: true, Busy: true},
128+ {Id: "r3", Name: "down"},
129+ }
130+ env := testEnv(t, map[string]string{"ERMITE_URL": f.serve(t), "ERMITE_TOKEN": "ermite_x"})
131+ env.now = func() time.Time { return time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC) } // a clock that is far off
132+ out := runCLIWith(t, env, "", "runner", "list").ok(t).stdout
133+ state := map[string]string{}
134+ for _, line := range strings.Split(strings.TrimSpace(out), "\n")[1:] {
135+ fields := strings.Fields(line)
136+ state[fields[1]] = fields[2]
137+ }
138+ assert.Equal(t, map[string]string{"up": "online", "busy": "busy", "down": "offline"}, state, out)
139+
140+ js := runCLIWith(t, env, "", "runner", "list", "--json").ok(t).stdout
141+ assert.Contains(t, js, `"online": true`)
142+}
@@ -4,7 +4,9 @@ package app
4 4
5 import (5 import (
6 "context"6 "context"
7+ "strings"
7 "testing"8 "testing"
9+ "time"
8 10
9 "connectrpc.com/connect"11 "connectrpc.com/connect"
10 "github.com/stretchr/testify/assert"12 "github.com/stretchr/testify/assert"
@@ -35,6 +37,9 @@ func (f *fakeAPI) RerunRun(_ context.Context, r *connect.Request[ermite.RerunRun
35 37
36 func (f *fakeAPI) ListRunners(_ context.Context, r *connect.Request[ermite.ListRunnersRequest]) (*connect.Response[ermite.ListRunnersResponse], error) {38 func (f *fakeAPI) ListRunners(_ context.Context, r *connect.Request[ermite.ListRunnersRequest]) (*connect.Response[ermite.ListRunnersResponse], error) {
37 return respond(f, "ListRunners", r, func(*ermite.ListRunnersRequest) (*ermite.ListRunnersResponse, error) {39 return respond(f, "ListRunners", r, func(*ermite.ListRunnersRequest) (*ermite.ListRunnersResponse, error) {
40+ if f.runners != nil {
41+ return &ermite.ListRunnersResponse{Runners: f.runners}, nil
42+ }
38 return &ermite.ListRunnersResponse{Runners: []*ermite.Runner{{Id: "r1", Name: "alpha"}}}, nil43 return &ermite.ListRunnersResponse{Runners: []*ermite.Runner{{Id: "r1", Name: "alpha"}}}, nil
39 })44 })
40 }45 }
@@ -112,3 +117,26 @@ func TestCICompletionNeverRunsTokenCmd(t *testing.T) {
112 })117 })
113 }118 }
114 }119 }
120+
121+// The CLI shows what the server decided: it never compares last_seen with its
122+// own clock.
123+func TestRunnerListShowsTheServersVerdict(t *testing.T) {
124+ f := &fakeAPI{}
125+ f.runners = []*ermite.Runner{
126+ {Id: "r1", Name: "up", Version: "v2", Online: true},
127+ {Id: "r2", Name: "busy", Online: true, Busy: true},
128+ {Id: "r3", Name: "down"},
129+ }
130+ env := testEnv(t, map[string]string{"ERMITE_URL": f.serve(t), "ERMITE_TOKEN": "ermite_x"})
131+ env.now = func() time.Time { return time.Date(2000, 1, 1, 0, 0, 0, 0, time.UTC) } // a clock that is far off
132+ out := runCLIWith(t, env, "", "runner", "list").ok(t).stdout
133+ state := map[string]string{}
134+ for _, line := range strings.Split(strings.TrimSpace(out), "\n")[1:] {
135+ fields := strings.Fields(line)
136+ state[fields[1]] = fields[2]
137+ }
138+ assert.Equal(t, map[string]string{"up": "online", "busy": "busy", "down": "offline"}, state, out)
139+
140+ js := runCLIWith(t, env, "", "runner", "list", "--json").ok(t).stdout
141+ assert.Contains(t, js, `"online": true`)
142+}
M internal/app/cli_fake_test.go +2 -0
@@ -34,6 +34,8 @@ type fakeAPI struct {
3434 ermiteconnect.UnimplementedSessionServiceHandler
3535 ermiteconnect.UnimplementedRunServiceHandler
3636
37+ runners []*ermite.Runner // ListRunners answer; nil means one default runner
38+
3739 mu sync.Mutex
3840 calls []string // procedure names, in order
3941 reqs []any // request messages, in order
@@ -34,6 +34,8 @@ type fakeAPI struct {
34 ermiteconnect.UnimplementedSessionServiceHandler34 ermiteconnect.UnimplementedSessionServiceHandler
35 ermiteconnect.UnimplementedRunServiceHandler35 ermiteconnect.UnimplementedRunServiceHandler
36 36
37+ runners []*ermite.Runner // ListRunners answer; nil means one default runner
38+
37 mu sync.Mutex39 mu sync.Mutex
38 calls []string // procedure names, in order40 calls []string // procedure names, in order
39 reqs []any // request messages, in order41 reqs []any // request messages, in order
M internal/app/cli_runner.go +5 -4
@@ -7,7 +7,6 @@ import (
77 "errors"
88 "fmt"
99 "strings"
10- "time"
1110
1211 "connectrpc.com/connect"
1312 "github.com/spf13/cobra"
@@ -27,12 +26,14 @@ type jsonRunner struct {
2726 RegisteredAt *string `json:"registered_at"`
2827 LastSeen *string `json:"last_seen"`
2928 Busy bool `json:"busy"`
29+ Online bool `json:"online"`
3030 }
3131
3232 const jsonRunnerHelp = `
3333 JSON fields of a runner (--json): id, name, labels, isolation (vm, container-sysbox,
3434 container-privileged), version, arch, os, registered_at, last_seen (RFC 3339 UTC, null if never
35-seen) and busy. A runner not seen for more than about 2 minutes is offline.`
35+seen), busy and online (decided by the server: heard from within the last 2 minutes by its
36+clock; a server older than the field reports false for every runner).`
3637
3738 func newRunnerCmd(env *cliEnv) *cobra.Command {
3839 runner := &cobra.Command{
@@ -79,7 +80,7 @@ func newRunnerListCmd(env *cliEnv) *cobra.Command {
7980 }
8081 out = append(out, jsonRunner{
8182 ID: r.GetId(), Name: r.GetName(), Labels: labels, Isolation: isolationText(r.GetIsolation()), Version: r.GetVersion(),
82- Arch: r.GetArch(), Os: r.GetOs(), RegisteredAt: jsonTime(r.GetRegisteredAt()), LastSeen: jsonTime(r.GetLastSeen()), Busy: r.GetBusy(),
83+ Arch: r.GetArch(), Os: r.GetOs(), RegisteredAt: jsonTime(r.GetRegisteredAt()), LastSeen: jsonTime(r.GetLastSeen()), Busy: r.GetBusy(), Online: r.GetOnline(),
8384 })
8485 }
8586 return writeJSON(cmd.OutOrStdout(), out)
@@ -91,7 +92,7 @@ func newRunnerListCmd(env *cliEnv) *cobra.Command {
9192 rows := make([][]string, 0, len(runners))
9293 for _, r := range runners {
9394 state := "offline"
94- if ls := r.GetLastSeen(); ls.IsValid() && env.now().Sub(ls.AsTime()) < 2*time.Minute {
95+ if r.GetOnline() {
9596 state = "online"
9697 if r.GetBusy() {
9798 state = "busy"
@@ -7,7 +7,6 @@ import (
7 "errors"7 "errors"
8 "fmt"8 "fmt"
9 "strings"9 "strings"
10- "time"
11 10
12 "connectrpc.com/connect"11 "connectrpc.com/connect"
13 "github.com/spf13/cobra"12 "github.com/spf13/cobra"
@@ -27,12 +26,14 @@ type jsonRunner struct {
27 RegisteredAt *string `json:"registered_at"`26 RegisteredAt *string `json:"registered_at"`
28 LastSeen *string `json:"last_seen"`27 LastSeen *string `json:"last_seen"`
29 Busy bool `json:"busy"`28 Busy bool `json:"busy"`
29+ Online bool `json:"online"`
30 }30 }
31 31
32 const jsonRunnerHelp = `32 const jsonRunnerHelp = `
33 JSON fields of a runner (--json): id, name, labels, isolation (vm, container-sysbox,33 JSON fields of a runner (--json): id, name, labels, isolation (vm, container-sysbox,
34 container-privileged), version, arch, os, registered_at, last_seen (RFC 3339 UTC, null if never34 container-privileged), version, arch, os, registered_at, last_seen (RFC 3339 UTC, null if never
35-seen) and busy. A runner not seen for more than about 2 minutes is offline.`35+seen), busy and online (decided by the server: heard from within the last 2 minutes by its
36+clock; a server older than the field reports false for every runner).`
36 37
37 func newRunnerCmd(env *cliEnv) *cobra.Command {38 func newRunnerCmd(env *cliEnv) *cobra.Command {
38 runner := &cobra.Command{39 runner := &cobra.Command{
@@ -79,7 +80,7 @@ func newRunnerListCmd(env *cliEnv) *cobra.Command {
79 }80 }
80 out = append(out, jsonRunner{81 out = append(out, jsonRunner{
81 ID: r.GetId(), Name: r.GetName(), Labels: labels, Isolation: isolationText(r.GetIsolation()), Version: r.GetVersion(),82 ID: r.GetId(), Name: r.GetName(), Labels: labels, Isolation: isolationText(r.GetIsolation()), Version: r.GetVersion(),
82- Arch: r.GetArch(), Os: r.GetOs(), RegisteredAt: jsonTime(r.GetRegisteredAt()), LastSeen: jsonTime(r.GetLastSeen()), Busy: r.GetBusy(),83+ Arch: r.GetArch(), Os: r.GetOs(), RegisteredAt: jsonTime(r.GetRegisteredAt()), LastSeen: jsonTime(r.GetLastSeen()), Busy: r.GetBusy(), Online: r.GetOnline(),
83 })84 })
84 }85 }
85 return writeJSON(cmd.OutOrStdout(), out)86 return writeJSON(cmd.OutOrStdout(), out)
@@ -91,7 +92,7 @@ func newRunnerListCmd(env *cliEnv) *cobra.Command {
91 rows := make([][]string, 0, len(runners))92 rows := make([][]string, 0, len(runners))
92 for _, r := range runners {93 for _, r := range runners {
93 state := "offline"94 state := "offline"
94- if ls := r.GetLastSeen(); ls.IsValid() && env.now().Sub(ls.AsTime()) < 2*time.Minute {95+ if r.GetOnline() {
95 state = "online"96 state = "online"
96 if r.GetBusy() {97 if r.GetBusy() {
97 state = "busy"98 state = "busy"
M internal/ports/ci.go +8 -0
@@ -172,6 +172,10 @@ type CIStore interface {
172172 FindRunnerByToken(ctx context.Context, hash domain.TokenHash) (domain.CIRunner, error)
173173 // ListRunners returns every runner ordered by name.
174174 ListRunners(ctx context.Context) ([]domain.CIRunner, error)
175+ // SetRunnerVersion records the version the runner runs now. It writes only
176+ // when the stored version differs, and returns domain.ErrNotFound for a
177+ // runner that is gone.
178+ SetRunnerVersion(ctx context.Context, id, version string) error
175179 // RemoveRunner removes the runner and its token, or returns
176180 // domain.ErrNotFound.
177181 RemoveRunner(ctx context.Context, id string) (domain.CIRunner, error)
@@ -462,6 +466,8 @@ func (j RunnerJob) GoString() string { return j.String() }
462466 type FetchJobParams struct {
463467 Labels []string
464468 PollID string
469+ // Version is the version of the running runner, "" from an older runner.
470+ Version string
465471 }
466472
467473 // FetchJobResult is the answer of a poll.
@@ -679,6 +685,8 @@ type RunnerView struct {
679685 domain.CIRunner
680686 LastSeen time.Time
681687 Busy bool
688+ // Online is decided by the server's clock (see usecase.RunnerOnlineWindow).
689+ Online bool
682690 }
683691
684692 // JobAnnotation is an annotation with the job it belongs to.
@@ -172,6 +172,10 @@ type CIStore interface {
172 FindRunnerByToken(ctx context.Context, hash domain.TokenHash) (domain.CIRunner, error)172 FindRunnerByToken(ctx context.Context, hash domain.TokenHash) (domain.CIRunner, error)
173 // ListRunners returns every runner ordered by name.173 // ListRunners returns every runner ordered by name.
174 ListRunners(ctx context.Context) ([]domain.CIRunner, error)174 ListRunners(ctx context.Context) ([]domain.CIRunner, error)
175+ // SetRunnerVersion records the version the runner runs now. It writes only
176+ // when the stored version differs, and returns domain.ErrNotFound for a
177+ // runner that is gone.
178+ SetRunnerVersion(ctx context.Context, id, version string) error
175 // RemoveRunner removes the runner and its token, or returns179 // RemoveRunner removes the runner and its token, or returns
176 // domain.ErrNotFound.180 // domain.ErrNotFound.
177 RemoveRunner(ctx context.Context, id string) (domain.CIRunner, error)181 RemoveRunner(ctx context.Context, id string) (domain.CIRunner, error)
@@ -462,6 +466,8 @@ func (j RunnerJob) GoString() string { return j.String() }
462 type FetchJobParams struct {466 type FetchJobParams struct {
463 Labels []string467 Labels []string
464 PollID string468 PollID string
469+ // Version is the version of the running runner, "" from an older runner.
470+ Version string
465 }471 }
466 472
467 // FetchJobResult is the answer of a poll.473 // FetchJobResult is the answer of a poll.
@@ -679,6 +685,8 @@ type RunnerView struct {
679 domain.CIRunner685 domain.CIRunner
680 LastSeen time.Time686 LastSeen time.Time
681 Busy bool687 Busy bool
688+ // Online is decided by the server's clock (see usecase.RunnerOnlineWindow).
689+ Online bool
682 }690 }
683 691
684 // JobAnnotation is an annotation with the job it belongs to.692 // JobAnnotation is an annotation with the job it belongs to.
M internal/ports/mocks/CIStore.go +63 -0
@@ -1535,6 +1535,69 @@ func (_c *CIStore_RunIDs_Call) RunAndReturn(run func(ctx context.Context, repo d
15351535 return _c
15361536 }
15371537
1538+// SetRunnerVersion provides a mock function for the type CIStore
1539+func (_mock *CIStore) SetRunnerVersion(ctx context.Context, id string, version string) error {
1540+ ret := _mock.Called(ctx, id, version)
1541+
1542+ if len(ret) == 0 {
1543+ panic("no return value specified for SetRunnerVersion")
1544+ }
1545+
1546+ var r0 error
1547+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, string) error); ok {
1548+ r0 = returnFunc(ctx, id, version)
1549+ } else {
1550+ r0 = ret.Error(0)
1551+ }
1552+ return r0
1553+}
1554+
1555+// CIStore_SetRunnerVersion_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'SetRunnerVersion'
1556+type CIStore_SetRunnerVersion_Call struct {
1557+ *mock.Call
1558+}
1559+
1560+// SetRunnerVersion is a helper method to define mock.On call
1561+// - ctx context.Context
1562+// - id string
1563+// - version string
1564+func (_e *CIStore_Expecter) SetRunnerVersion(ctx any, id any, version any) *CIStore_SetRunnerVersion_Call {
1565+ return &CIStore_SetRunnerVersion_Call{Call: _e.mock.On("SetRunnerVersion", ctx, id, version)}
1566+}
1567+
1568+func (_c *CIStore_SetRunnerVersion_Call) Run(run func(ctx context.Context, id string, version string)) *CIStore_SetRunnerVersion_Call {
1569+ _c.Call.Run(func(args mock.Arguments) {
1570+ var arg0 context.Context
1571+ if args[0] != nil {
1572+ arg0 = args[0].(context.Context)
1573+ }
1574+ var arg1 string
1575+ if args[1] != nil {
1576+ arg1 = args[1].(string)
1577+ }
1578+ var arg2 string
1579+ if args[2] != nil {
1580+ arg2 = args[2].(string)
1581+ }
1582+ run(
1583+ arg0,
1584+ arg1,
1585+ arg2,
1586+ )
1587+ })
1588+ return _c
1589+}
1590+
1591+func (_c *CIStore_SetRunnerVersion_Call) Return(err error) *CIStore_SetRunnerVersion_Call {
1592+ _c.Call.Return(err)
1593+ return _c
1594+}
1595+
1596+func (_c *CIStore_SetRunnerVersion_Call) RunAndReturn(run func(ctx context.Context, id string, version string) error) *CIStore_SetRunnerVersion_Call {
1597+ _c.Call.Return(run)
1598+ return _c
1599+}
1600+
15381601 // Sweep provides a mock function for the type CIStore
15391602 func (_mock *CIStore) Sweep(ctx context.Context, before time.Time) ([]int64, error) {
15401603 ret := _mock.Called(ctx, before)
@@ -1535,6 +1535,69 @@ func (_c *CIStore_RunIDs_Call) RunAndReturn(run func(ctx context.Context, repo d
1535 return _c1535 return _c
1536 }1536 }
1537 1537
1538+// SetRunnerVersion provides a mock function for the type CIStore
1539+func (_mock *CIStore) SetRunnerVersion(ctx context.Context, id string, version string) error {
1540+ ret := _mock.Called(ctx, id, version)
1541+
1542+ if len(ret) == 0 {
1543+ panic("no return value specified for SetRunnerVersion")
1544+ }
1545+
1546+ var r0 error
1547+ if returnFunc, ok := ret.Get(0).(func(context.Context, string, string) error); ok {
1548+ r0 = returnFunc(ctx, id, version)
1549+ } else {
1550+ r0 = ret.Error(0)
1551+ }
1552+ return r0
1553+}
1554+
1555+// CIStore_SetRunnerVersion_Call is a *mock.Call that shadows Run/Return methods with type explicit version for method 'SetRunnerVersion'
1556+type CIStore_SetRunnerVersion_Call struct {
1557+ *mock.Call
1558+}
1559+
1560+// SetRunnerVersion is a helper method to define mock.On call
1561+// - ctx context.Context
1562+// - id string
1563+// - version string
1564+func (_e *CIStore_Expecter) SetRunnerVersion(ctx any, id any, version any) *CIStore_SetRunnerVersion_Call {
1565+ return &CIStore_SetRunnerVersion_Call{Call: _e.mock.On("SetRunnerVersion", ctx, id, version)}
1566+}
1567+
1568+func (_c *CIStore_SetRunnerVersion_Call) Run(run func(ctx context.Context, id string, version string)) *CIStore_SetRunnerVersion_Call {
1569+ _c.Call.Run(func(args mock.Arguments) {
1570+ var arg0 context.Context
1571+ if args[0] != nil {
1572+ arg0 = args[0].(context.Context)
1573+ }
1574+ var arg1 string
1575+ if args[1] != nil {
1576+ arg1 = args[1].(string)
1577+ }
1578+ var arg2 string
1579+ if args[2] != nil {
1580+ arg2 = args[2].(string)
1581+ }
1582+ run(
1583+ arg0,
1584+ arg1,
1585+ arg2,
1586+ )
1587+ })
1588+ return _c
1589+}
1590+
1591+func (_c *CIStore_SetRunnerVersion_Call) Return(err error) *CIStore_SetRunnerVersion_Call {
1592+ _c.Call.Return(err)
1593+ return _c
1594+}
1595+
1596+func (_c *CIStore_SetRunnerVersion_Call) RunAndReturn(run func(ctx context.Context, id string, version string) error) *CIStore_SetRunnerVersion_Call {
1597+ _c.Call.Return(run)
1598+ return _c
1599+}
1600+
1538 // Sweep provides a mock function for the type CIStore1601 // Sweep provides a mock function for the type CIStore
1539 func (_mock *CIStore) Sweep(ctx context.Context, before time.Time) ([]int64, error) {1602 func (_mock *CIStore) Sweep(ctx context.Context, before time.Time) ([]int64, error) {
1540 ret := _mock.Called(ctx, before)1603 ret := _mock.Called(ctx, before)
M internal/usecase/ci_runner.go +26 -0
@@ -97,6 +97,28 @@ func (c *CI) runnerFor(ctx context.Context, secret string) (domain.CIRunner, err
9797 return r, err
9898 }
9999
100+// RunnerOnlineWindow is how long after its last sign of life (a poll or a
101+// heartbeat) a runner is online, by the server's clock.
102+const RunnerOnlineWindow = 2 * time.Minute
103+
104+func (c *CI) runnerOnline(lastSeen, now time.Time) bool {
105+ return !lastSeen.IsZero() && now.Sub(lastSeen) < RunnerOnlineWindow
106+}
107+
108+// noteVersion records the version a poll reports, once: the runner read for
109+// this poll already holds the stored one, so an unchanged version costs no
110+// write. A runner that sends none keeps its stored version. A failure to
111+// record is logged and does not fail the poll.
112+func (c *CI) noteVersion(ctx context.Context, r domain.CIRunner, version string) {
113+ if version == "" || version == r.Version {
114+ return
115+ }
116+ if err := c.cfg.Store.SetRunnerVersion(ctx, r.ID, version); err != nil && !errors.Is(err, domain.ErrNotFound) {
117+ log := zerowrap.FromCtx(ctx)
118+ log.Warn().Err(err).Str("runner_id", r.ID).Msg("runner version not recorded")
119+ }
120+}
121+
100122 func (c *CI) markSeen(runnerID string) {
101123 c.mu.Lock()
102124 c.seen[runnerID] = c.now()
@@ -115,6 +137,9 @@ func (c *CI) FetchJob(ctx context.Context, secret string, params ports.FetchJobP
115137 if err != nil {
116138 return ports.FetchJobResult{}, err
117139 }
140+ if len(params.Version) > maxVersionLen || !utf8.ValidString(params.Version) {
141+ return ports.FetchJobResult{}, fmt.Errorf("%w: version is at most 64 characters", domain.ErrInvalid)
142+ }
118143 offered, err := offeredLabels(runner, params.Labels)
119144 if err != nil {
120145 return ports.FetchJobResult{}, err
@@ -132,6 +157,7 @@ func (c *CI) FetchJob(ctx context.Context, secret string, params ports.FetchJobP
132157 }
133158 }
134159 c.markSeen(runner.ID)
160+ c.noteVersion(ctx, runner, params.Version)
135161 deadline := time.NewTimer(c.cfg.PollTimeout)
136162 defer deadline.Stop()
137163 for {
@@ -97,6 +97,28 @@ func (c *CI) runnerFor(ctx context.Context, secret string) (domain.CIRunner, err
97 return r, err97 return r, err
98 }98 }
99 99
100+// RunnerOnlineWindow is how long after its last sign of life (a poll or a
101+// heartbeat) a runner is online, by the server's clock.
102+const RunnerOnlineWindow = 2 * time.Minute
103+
104+func (c *CI) runnerOnline(lastSeen, now time.Time) bool {
105+ return !lastSeen.IsZero() && now.Sub(lastSeen) < RunnerOnlineWindow
106+}
107+
108+// noteVersion records the version a poll reports, once: the runner read for
109+// this poll already holds the stored one, so an unchanged version costs no
110+// write. A runner that sends none keeps its stored version. A failure to
111+// record is logged and does not fail the poll.
112+func (c *CI) noteVersion(ctx context.Context, r domain.CIRunner, version string) {
113+ if version == "" || version == r.Version {
114+ return
115+ }
116+ if err := c.cfg.Store.SetRunnerVersion(ctx, r.ID, version); err != nil && !errors.Is(err, domain.ErrNotFound) {
117+ log := zerowrap.FromCtx(ctx)
118+ log.Warn().Err(err).Str("runner_id", r.ID).Msg("runner version not recorded")
119+ }
120+}
121+
100 func (c *CI) markSeen(runnerID string) {122 func (c *CI) markSeen(runnerID string) {
101 c.mu.Lock()123 c.mu.Lock()
102 c.seen[runnerID] = c.now()124 c.seen[runnerID] = c.now()
@@ -115,6 +137,9 @@ func (c *CI) FetchJob(ctx context.Context, secret string, params ports.FetchJobP
115 if err != nil {137 if err != nil {
116 return ports.FetchJobResult{}, err138 return ports.FetchJobResult{}, err
117 }139 }
140+ if len(params.Version) > maxVersionLen || !utf8.ValidString(params.Version) {
141+ return ports.FetchJobResult{}, fmt.Errorf("%w: version is at most 64 characters", domain.ErrInvalid)
142+ }
118 offered, err := offeredLabels(runner, params.Labels)143 offered, err := offeredLabels(runner, params.Labels)
119 if err != nil {144 if err != nil {
120 return ports.FetchJobResult{}, err145 return ports.FetchJobResult{}, err
@@ -132,6 +157,7 @@ func (c *CI) FetchJob(ctx context.Context, secret string, params ports.FetchJobP
132 }157 }
133 }158 }
134 c.markSeen(runner.ID)159 c.markSeen(runner.ID)
160+ c.noteVersion(ctx, runner, params.Version)
135 deadline := time.NewTimer(c.cfg.PollTimeout)161 deadline := time.NewTimer(c.cfg.PollTimeout)
136 defer deadline.Stop()162 defer deadline.Stop()
137 for {163 for {
M internal/usecase/ci_runs.go +2 -1
@@ -584,10 +584,11 @@ func (c *CI) ListRunners(ctx context.Context, p domain.Principal, size int, toke
584584 func(r domain.CIRunner) string {
585585 return base64.RawURLEncoding.EncodeToString([]byte("rn1:" + strings.ToLower(r.Name)))
586586 })
587+ now := c.now()
587588 c.mu.Lock()
588589 defer c.mu.Unlock()
589590 for _, r := range rows {
590- page.Runners = append(page.Runners, ports.RunnerView{CIRunner: r, LastSeen: c.seen[r.ID], Busy: busy[r.ID]})
591+ page.Runners = append(page.Runners, ports.RunnerView{CIRunner: r, LastSeen: c.seen[r.ID], Busy: busy[r.ID], Online: c.runnerOnline(c.seen[r.ID], now)})
591592 }
592593 return page, nil
593594 }
@@ -584,10 +584,11 @@ func (c *CI) ListRunners(ctx context.Context, p domain.Principal, size int, toke
584 func(r domain.CIRunner) string {584 func(r domain.CIRunner) string {
585 return base64.RawURLEncoding.EncodeToString([]byte("rn1:" + strings.ToLower(r.Name)))585 return base64.RawURLEncoding.EncodeToString([]byte("rn1:" + strings.ToLower(r.Name)))
586 })586 })
587+ now := c.now()
587 c.mu.Lock()588 c.mu.Lock()
588 defer c.mu.Unlock()589 defer c.mu.Unlock()
589 for _, r := range rows {590 for _, r := range rows {
590- page.Runners = append(page.Runners, ports.RunnerView{CIRunner: r, LastSeen: c.seen[r.ID], Busy: busy[r.ID]})591+ page.Runners = append(page.Runners, ports.RunnerView{CIRunner: r, LastSeen: c.seen[r.ID], Busy: busy[r.ID], Online: c.runnerOnline(c.seen[r.ID], now)})
591 }592 }
592 return page, nil593 return page, nil
593 }594 }

Keyboard shortcuts

Single-key shortcuts are on. ctrl k and esc work whatever this says.

ctrl k :
go to a section, branch, tag, file or repository
?
this list
j k
move the row cursor down / up
enter
open the row under the cursor; on Files, a directory opens or closes in place
↑ ↓ ← → home end
in the file tree (focus on a row): previous / next row, close / open a directory or go to its parent / first entry, first / last row. They never open a file
l h
on Files: open a directory or a file / close a directory or go up to its parent, then to the parent page. Elsewhere: into the side panel (on a merge request, its files) / back to the content; there j k move between its controls
g g G
first / last row
/
filter the list; on Files, find a file anywhere in the repository (esc clears, enter opens the best match)
b
switch branch or tag
g s g f g c g b g m g i g r g a
tabs: summary, files, commits, branches & tags, merge requests, issues, releases, actions (when available)
g h
home
w
wide mode: use the whole width of the screen, or not (remembered)
y
copy a permalink pinned to the commit
d u x
diff: split / unified / collapse or expand all
e E
diff: the whole file in view / every file, with the lines the hunks hide
n
merge request: next note in the diff
esc
close a dialog, leave a field

Branches 2

Tags 42

↑↓ moveenter openesc close