Monorepo for Tangled
spindle/mill: authenticate executors with per-executor tokens
Registers executors in the db and resolves identity from the token hash
in the Authorization header rather than the client-supplied Hello. Adds
`spindle mill executor add/list/revoke` and seeds the dev SharedSecret as
a bootstrap identity.
+869
-133
+3
spindle/server.go
+3
spindle/server.go
···
420
420
// but the mill's db/notifier are created inside New.
421
421
m.Attach(s.DB(), s.Notifier())
422
422
s.mill = m
423
+
if err := m.SeedBootstrapToken(); err != nil {
424
+
return fmt.Errorf("seeding mill bootstrap token: %w", err)
425
+
}
423
426
}
424
427
if cfg.Role == config.RoleExecutor {
425
428
s.exec = executor.New(cfg, engines, s.DB(), s.Notifier(), log.SubLogger(logger, "executor"))
+120
cmd/spindle/main.go
+120
cmd/spindle/main.go
···
2
2
3
3
import (
4
4
"context"
5
+
"fmt"
5
6
"log/slog"
6
7
"os"
8
+
"text/tabwriter"
9
+
"time"
7
10
8
11
"github.com/urfave/cli/v3"
9
12
tlog "tangled.org/core/log"
10
13
"tangled.org/core/spindle"
14
+
"tangled.org/core/spindle/db"
15
+
"tangled.org/core/spindle/mill"
11
16
)
12
17
13
18
func main() {
···
16
21
Usage: "spindle continuous integration runner",
17
22
Commands: []*cli.Command{
18
23
Command(),
24
+
millCommand(),
19
25
},
20
26
DefaultCommand: "run",
21
27
}
···
38
44
Usage: "run the spindle server",
39
45
Action: func(ctx context.Context, cmd *cli.Command) error {
40
46
return spindle.Run(ctx)
47
+
},
48
+
}
49
+
}
50
+
51
+
// millCommand groups mill-host administration. Executors allowed to join the
52
+
// mill are managed under `mill executor`.
53
+
func millCommand() *cli.Command {
54
+
dbFlag := &cli.StringFlag{
55
+
Name: "db",
56
+
Usage: "path to the spindle sqlite db",
57
+
Value: "spindle.db",
58
+
Sources: cli.EnvVars("SPINDLE_SERVER_DB_PATH"),
59
+
}
60
+
openDB := func(ctx context.Context, cmd *cli.Command) (*db.DB, error) {
61
+
return db.Make(ctx, cmd.String("db"))
62
+
}
63
+
return &cli.Command{
64
+
Name: "mill",
65
+
Usage: "mill host administration",
66
+
Commands: []*cli.Command{
67
+
{
68
+
Name: "executor",
69
+
Usage: "manage executors allowed to join this mill",
70
+
Commands: []*cli.Command{
71
+
{
72
+
Name: "add",
73
+
Usage: "register an executor and print its token",
74
+
ArgsUsage: "<name>",
75
+
Flags: []cli.Flag{
76
+
dbFlag,
77
+
&cli.DurationFlag{
78
+
Name: "ttl",
79
+
Usage: "token lifetime (e.g. 720h); omit for no expiry",
80
+
},
81
+
},
82
+
Action: func(ctx context.Context, cmd *cli.Command) error {
83
+
name := cmd.Args().First()
84
+
if name == "" {
85
+
return fmt.Errorf("usage: spindle mill executor add <name>")
86
+
}
87
+
d, err := openDB(ctx, cmd)
88
+
if err != nil {
89
+
return err
90
+
}
91
+
token, err := mill.GenerateToken()
92
+
if err != nil {
93
+
return err
94
+
}
95
+
var expiresAt *time.Time
96
+
if ttl := cmd.Duration("ttl"); ttl > 0 {
97
+
exp := time.Now().UTC().Add(ttl)
98
+
expiresAt = &exp
99
+
}
100
+
if err := d.AddExecutorToken(name, mill.HashToken(token), expiresAt); err != nil {
101
+
return fmt.Errorf("registering executor %q: %w", name, err)
102
+
}
103
+
fmt.Println(token)
104
+
return nil
105
+
},
106
+
},
107
+
{
108
+
Name: "list",
109
+
Usage: "list registered executors",
110
+
Flags: []cli.Flag{dbFlag},
111
+
Action: func(ctx context.Context, cmd *cli.Command) error {
112
+
d, err := openDB(ctx, cmd)
113
+
if err != nil {
114
+
return err
115
+
}
116
+
tokens, err := d.ListExecutorTokens()
117
+
if err != nil {
118
+
return err
119
+
}
120
+
w := tabwriter.NewWriter(os.Stdout, 0, 0, 3, ' ', 0)
121
+
fmt.Fprintln(w, "NAME\tCREATED\tEXPIRES")
122
+
for _, t := range tokens {
123
+
expires := "never"
124
+
if t.ExpiresAt != nil {
125
+
expires = t.ExpiresAt.Format(time.RFC3339)
126
+
if time.Now().After(*t.ExpiresAt) {
127
+
expires += " (expired)"
128
+
}
129
+
}
130
+
fmt.Fprintf(w, "%s\t%s\t%s\n", t.Name, t.CreatedAt, expires)
131
+
}
132
+
return w.Flush()
133
+
},
134
+
},
135
+
{
136
+
Name: "revoke",
137
+
Usage: "revoke an executor's token",
138
+
ArgsUsage: "<name>",
139
+
Flags: []cli.Flag{dbFlag},
140
+
Action: func(ctx context.Context, cmd *cli.Command) error {
141
+
name := cmd.Args().First()
142
+
if name == "" {
143
+
return fmt.Errorf("usage: spindle mill executor revoke <name>")
144
+
}
145
+
d, err := openDB(ctx, cmd)
146
+
if err != nil {
147
+
return err
148
+
}
149
+
ok, err := d.RevokeExecutorToken(name)
150
+
if err != nil {
151
+
return err
152
+
}
153
+
if !ok {
154
+
return fmt.Errorf("no such executor identity %q", name)
155
+
}
156
+
return nil
157
+
},
158
+
},
159
+
},
160
+
},
41
161
},
42
162
}
43
163
}
+3
spindle/config/config.go
+3
spindle/config/config.go
···
133
133
if c.Mill.URL == "" {
134
134
return fmt.Errorf("SPINDLE_ROLE=executor requires SPINDLE_MILL_URL (the mill to dial)")
135
135
}
136
+
if c.Mill.SharedSecret == "" {
137
+
return fmt.Errorf("SPINDLE_ROLE=executor requires SPINDLE_MILL_SHARED_SECRET (its executor token)")
138
+
}
136
139
default:
137
140
return fmt.Errorf("unknown SPINDLE_ROLE %q (want standalone, mill, or executor)", c.Role)
138
141
}
+7
spindle/db/db.go
+7
spindle/db/db.go
···
123
123
foreign key (pipeline_id) references pipelines(id) on delete cascade
124
124
);
125
125
126
+
create table if not exists mill_executors (
127
+
name text primary key,
128
+
token_hash text not null,
129
+
created_at text not null default (strftime('%Y-%m-%dT%H:%M:%SZ', 'now')),
130
+
expires_at text
131
+
);
132
+
126
133
create table if not exists migrations (
127
134
id integer primary key autoincrement,
128
135
name text unique
+111
spindle/db/mill_tokens.go
+111
spindle/db/mill_tokens.go
···
1
+
package db
2
+
3
+
import (
4
+
"database/sql"
5
+
"time"
6
+
)
7
+
8
+
// ExecutorToken is a registered mill executor identity. The raw token is never
9
+
// stored; token_hash is a hash of it (see mill.HashToken). ExpiresAt is nil for
10
+
// a token that never expires.
11
+
type ExecutorToken struct {
12
+
Name string
13
+
CreatedAt string
14
+
ExpiresAt *time.Time
15
+
}
16
+
17
+
// AddExecutorToken registers a new executor identity. expiresAt is optional (nil
18
+
// means the token never expires). It fails if the name is already taken.
19
+
func (d *DB) AddExecutorToken(name, tokenHash string, expiresAt *time.Time) error {
20
+
_, err := d.Exec(
21
+
`insert into mill_executors (name, token_hash, expires_at) values (?, ?, ?)`,
22
+
name, tokenHash, expiryArg(expiresAt),
23
+
)
24
+
return err
25
+
}
26
+
27
+
// UpsertExecutorToken registers or replaces an executor identity's token with no
28
+
// expiry. Used for the dev bootstrap seed; production registers per-executor
29
+
// tokens via AddExecutorToken.
30
+
func (d *DB) UpsertExecutorToken(name, tokenHash string) error {
31
+
_, err := d.Exec(
32
+
`insert into mill_executors (name, token_hash, expires_at) values (?, ?, null)
33
+
on conflict(name) do update set token_hash = excluded.token_hash, expires_at = null`,
34
+
name, tokenHash,
35
+
)
36
+
return err
37
+
}
38
+
39
+
// ResolveExecutorToken returns the executor name a token hash is registered
40
+
// under. ok is false when no identity matches or the token has expired.
41
+
func (d *DB) ResolveExecutorToken(tokenHash string) (string, bool, error) {
42
+
var name string
43
+
var expires sql.NullString
44
+
err := d.QueryRow(
45
+
`select name, expires_at from mill_executors where token_hash = ?`, tokenHash,
46
+
).Scan(&name, &expires)
47
+
if err == sql.ErrNoRows {
48
+
return "", false, nil
49
+
}
50
+
if err != nil {
51
+
return "", false, err
52
+
}
53
+
if exp, ok := parseExpiry(expires); ok && time.Now().After(exp) {
54
+
return "", false, nil
55
+
}
56
+
return name, true, nil
57
+
}
58
+
59
+
// RevokeExecutorToken removes an executor identity, returning whether a row was
60
+
// deleted.
61
+
func (d *DB) RevokeExecutorToken(name string) (bool, error) {
62
+
res, err := d.Exec(`delete from mill_executors where name = ?`, name)
63
+
if err != nil {
64
+
return false, err
65
+
}
66
+
n, err := res.RowsAffected()
67
+
return n > 0, err
68
+
}
69
+
70
+
// ListExecutorTokens returns the registered executor identities ordered by name.
71
+
func (d *DB) ListExecutorTokens() ([]ExecutorToken, error) {
72
+
rows, err := d.Query(`select name, created_at, expires_at from mill_executors order by name`)
73
+
if err != nil {
74
+
return nil, err
75
+
}
76
+
defer rows.Close()
77
+
78
+
var out []ExecutorToken
79
+
for rows.Next() {
80
+
var t ExecutorToken
81
+
var expires sql.NullString
82
+
if err := rows.Scan(&t.Name, &t.CreatedAt, &expires); err != nil {
83
+
return nil, err
84
+
}
85
+
if exp, ok := parseExpiry(expires); ok {
86
+
t.ExpiresAt = &exp
87
+
}
88
+
out = append(out, t)
89
+
}
90
+
return out, rows.Err()
91
+
}
92
+
93
+
// expiryArg renders an optional expiry as a sql argument (nil -> NULL).
94
+
func expiryArg(t *time.Time) any {
95
+
if t == nil {
96
+
return nil
97
+
}
98
+
return t.UTC().Format(time.RFC3339)
99
+
}
100
+
101
+
// parseExpiry decodes a stored expiry column; ok is false for NULL or unparseable.
102
+
func parseExpiry(s sql.NullString) (time.Time, bool) {
103
+
if !s.Valid || s.String == "" {
104
+
return time.Time{}, false
105
+
}
106
+
t, err := time.Parse(time.RFC3339, s.String)
107
+
if err != nil {
108
+
return time.Time{}, false
109
+
}
110
+
return t, true
111
+
}
+235
spindle/db/mill_tokens_test.go
+235
spindle/db/mill_tokens_test.go
···
1
+
package db
2
+
3
+
import (
4
+
"testing"
5
+
"time"
6
+
)
7
+
8
+
func TestAddExecutorTokenRejectsDuplicateName(t *testing.T) {
9
+
d := newTestDB(t)
10
+
11
+
if err := d.AddExecutorToken("exec-1", "hash-a", nil); err != nil {
12
+
t.Fatalf("AddExecutorToken: %v", err)
13
+
}
14
+
if err := d.AddExecutorToken("exec-1", "hash-b", nil); err == nil {
15
+
t.Fatal("AddExecutorToken re-registered an existing name; a duplicate must be rejected")
16
+
}
17
+
18
+
name, ok, err := d.ResolveExecutorToken("hash-a")
19
+
if err != nil {
20
+
t.Fatalf("ResolveExecutorToken(hash-a): %v", err)
21
+
}
22
+
if !ok || name != "exec-1" {
23
+
t.Fatalf("ResolveExecutorToken(hash-a) = (%q, %v), want (exec-1, true)", name, ok)
24
+
}
25
+
if _, ok, _ := d.ResolveExecutorToken("hash-b"); ok {
26
+
t.Fatal("rejected duplicate's token resolved; the failed insert leaked a credential")
27
+
}
28
+
}
29
+
30
+
func TestResolveExecutorTokenMissAndHit(t *testing.T) {
31
+
d := newTestDB(t)
32
+
33
+
name, ok, err := d.ResolveExecutorToken("no-such-hash")
34
+
if err != nil {
35
+
t.Fatalf("ResolveExecutorToken(miss): %v", err)
36
+
}
37
+
if ok || name != "" {
38
+
t.Fatalf("ResolveExecutorToken(miss) = (%q, %v), want (\"\", false)", name, ok)
39
+
}
40
+
41
+
if err := d.AddExecutorToken("exec-1", "hash-1", nil); err != nil {
42
+
t.Fatalf("AddExecutorToken: %v", err)
43
+
}
44
+
45
+
name, ok, err = d.ResolveExecutorToken("hash-1")
46
+
if err != nil {
47
+
t.Fatalf("ResolveExecutorToken(hit): %v", err)
48
+
}
49
+
if !ok || name != "exec-1" {
50
+
t.Fatalf("ResolveExecutorToken(hash-1) = (%q, %v), want (exec-1, true)", name, ok)
51
+
}
52
+
53
+
if _, ok, _ := d.ResolveExecutorToken("hash-unregistered"); ok {
54
+
t.Fatal("ResolveExecutorToken matched an unregistered hash")
55
+
}
56
+
}
57
+
58
+
func TestUpsertExecutorTokenRotatesHash(t *testing.T) {
59
+
d := newTestDB(t)
60
+
61
+
if err := d.AddExecutorToken("exec-1", "old-hash", nil); err != nil {
62
+
t.Fatalf("AddExecutorToken: %v", err)
63
+
}
64
+
if err := d.UpsertExecutorToken("exec-1", "new-hash"); err != nil {
65
+
t.Fatalf("UpsertExecutorToken(rotate): %v", err)
66
+
}
67
+
68
+
name, ok, err := d.ResolveExecutorToken("new-hash")
69
+
if err != nil {
70
+
t.Fatalf("ResolveExecutorToken(new-hash): %v", err)
71
+
}
72
+
if !ok || name != "exec-1" {
73
+
t.Fatalf("ResolveExecutorToken(new-hash) = (%q, %v), want (exec-1, true)", name, ok)
74
+
}
75
+
if _, ok, _ := d.ResolveExecutorToken("old-hash"); ok {
76
+
t.Fatal("rotated-out token still resolves; UpsertExecutorToken did not replace the hash")
77
+
}
78
+
79
+
if err := d.UpsertExecutorToken("exec-2", "hash-2"); err != nil {
80
+
t.Fatalf("UpsertExecutorToken(insert): %v", err)
81
+
}
82
+
name, ok, err = d.ResolveExecutorToken("hash-2")
83
+
if err != nil {
84
+
t.Fatalf("ResolveExecutorToken(hash-2): %v", err)
85
+
}
86
+
if !ok || name != "exec-2" {
87
+
t.Fatalf("ResolveExecutorToken(hash-2) = (%q, %v), want (exec-2, true)", name, ok)
88
+
}
89
+
}
90
+
91
+
func TestRevokeExecutorToken(t *testing.T) {
92
+
d := newTestDB(t)
93
+
94
+
if err := d.AddExecutorToken("exec-1", "hash-1", nil); err != nil {
95
+
t.Fatalf("AddExecutorToken: %v", err)
96
+
}
97
+
98
+
deleted, err := d.RevokeExecutorToken("exec-1")
99
+
if err != nil {
100
+
t.Fatalf("RevokeExecutorToken: %v", err)
101
+
}
102
+
if !deleted {
103
+
t.Fatal("RevokeExecutorToken reported no deletion for an existing identity")
104
+
}
105
+
106
+
if _, ok, _ := d.ResolveExecutorToken("hash-1"); ok {
107
+
t.Fatal("revoked token still resolves; revocation is not enforced")
108
+
}
109
+
110
+
if deleted, err := d.RevokeExecutorToken("exec-1"); err != nil || deleted {
111
+
t.Fatalf("RevokeExecutorToken(already-gone) = (%v, %v), want (false, nil)", deleted, err)
112
+
}
113
+
114
+
if deleted, err := d.RevokeExecutorToken("ghost"); err != nil || deleted {
115
+
t.Fatalf("RevokeExecutorToken(unknown) = (%v, %v), want (false, nil)", deleted, err)
116
+
}
117
+
}
118
+
119
+
func TestListExecutorTokens(t *testing.T) {
120
+
d := newTestDB(t)
121
+
122
+
tokens, err := d.ListExecutorTokens()
123
+
if err != nil {
124
+
t.Fatalf("ListExecutorTokens(empty): %v", err)
125
+
}
126
+
if len(tokens) != 0 {
127
+
t.Fatalf("ListExecutorTokens on empty table = %d rows, want 0", len(tokens))
128
+
}
129
+
130
+
// insert out of alphabetical order to prove the ordering is the query's.
131
+
for _, name := range []string{"charlie", "alice", "bob"} {
132
+
if err := d.AddExecutorToken(name, "hash-"+name, nil); err != nil {
133
+
t.Fatalf("AddExecutorToken(%s): %v", name, err)
134
+
}
135
+
}
136
+
137
+
tokens, err = d.ListExecutorTokens()
138
+
if err != nil {
139
+
t.Fatalf("ListExecutorTokens: %v", err)
140
+
}
141
+
want := []string{"alice", "bob", "charlie"}
142
+
if len(tokens) != len(want) {
143
+
t.Fatalf("ListExecutorTokens = %d rows, want %d", len(tokens), len(want))
144
+
}
145
+
for i := range want {
146
+
if tokens[i].Name != want[i] {
147
+
t.Fatalf("ListExecutorTokens[%d].Name = %q, want %q (ordered by name)", i, tokens[i].Name, want[i])
148
+
}
149
+
}
150
+
}
151
+
152
+
// expired tokens must fail closed (ok=false, no error).
153
+
func TestResolveExecutorTokenExpiry(t *testing.T) {
154
+
cases := []struct {
155
+
name string
156
+
offset time.Duration
157
+
noExpiry bool
158
+
wantOK bool
159
+
}{
160
+
{"future expiry resolves", time.Hour, false, true},
161
+
{"past expiry fails closed", -time.Hour, false, false},
162
+
{"nil expiry never expires", 0, true, true},
163
+
}
164
+
for _, tc := range cases {
165
+
t.Run(tc.name, func(t *testing.T) {
166
+
d := newTestDB(t)
167
+
168
+
var expires *time.Time
169
+
if !tc.noExpiry {
170
+
exp := time.Now().Add(tc.offset)
171
+
expires = &exp
172
+
}
173
+
if err := d.AddExecutorToken("exec-1", "hash-1", expires); err != nil {
174
+
t.Fatalf("AddExecutorToken: %v", err)
175
+
}
176
+
177
+
name, ok, err := d.ResolveExecutorToken("hash-1")
178
+
if err != nil {
179
+
t.Fatalf("ResolveExecutorToken: %v", err)
180
+
}
181
+
if ok != tc.wantOK {
182
+
t.Fatalf("ResolveExecutorToken ok = %v, want %v", ok, tc.wantOK)
183
+
}
184
+
if tc.wantOK && name != "exec-1" {
185
+
t.Fatalf("ResolveExecutorToken name = %q, want exec-1", name)
186
+
}
187
+
if !tc.wantOK && name != "" {
188
+
t.Fatalf("ResolveExecutorToken name = %q, want \"\" when failing closed", name)
189
+
}
190
+
})
191
+
}
192
+
}
193
+
194
+
// expiry storage is RFC3339 second-precision UTC, so round-trips compare to
195
+
// the second.
196
+
func TestListExecutorTokensSurfacesExpiry(t *testing.T) {
197
+
d := newTestDB(t)
198
+
199
+
exp := time.Now().Add(24 * time.Hour)
200
+
if err := d.AddExecutorToken("expiring", "hash-exp", &exp); err != nil {
201
+
t.Fatalf("AddExecutorToken(expiring): %v", err)
202
+
}
203
+
if err := d.AddExecutorToken("forever", "hash-forever", nil); err != nil {
204
+
t.Fatalf("AddExecutorToken(forever): %v", err)
205
+
}
206
+
207
+
tokens, err := d.ListExecutorTokens()
208
+
if err != nil {
209
+
t.Fatalf("ListExecutorTokens: %v", err)
210
+
}
211
+
212
+
got := make(map[string]*time.Time, len(tokens))
213
+
for _, tok := range tokens {
214
+
got[tok.Name] = tok.ExpiresAt
215
+
}
216
+
217
+
e, present := got["forever"]
218
+
if !present {
219
+
t.Fatal("ListExecutorTokens omitted the non-expiring identity")
220
+
}
221
+
if e != nil {
222
+
t.Fatalf("forever.ExpiresAt = %v, want nil (never expires)", e)
223
+
}
224
+
225
+
e, present = got["expiring"]
226
+
if !present {
227
+
t.Fatal("ListExecutorTokens omitted the expiring identity")
228
+
}
229
+
if e == nil {
230
+
t.Fatal("expiring.ExpiresAt = nil, want the stored expiry")
231
+
}
232
+
if e.Unix() != exp.Unix() {
233
+
t.Fatalf("expiring.ExpiresAt = %d (unix), want %d", e.Unix(), exp.Unix())
234
+
}
235
+
}
+32
spindle/mill/auth_test.go
+32
spindle/mill/auth_test.go
···
25
25
return scriptedEncoder(func(*millproto.Message) error { return nil })
26
26
}
27
27
28
+
func TestHashToken(t *testing.T) {
29
+
const raw = "super-secret-executor-token"
30
+
31
+
if HashToken(raw) != HashToken(raw) {
32
+
t.Fatal("HashToken is not deterministic; the same token would stop authenticating")
33
+
}
34
+
if HashToken("token-a") == HashToken("token-b") {
35
+
t.Fatal("HashToken collided two distinct tokens")
36
+
}
37
+
if HashToken(raw) == raw {
38
+
t.Fatal("HashToken returned the raw token; a hash leak would expose a usable credential")
39
+
}
40
+
}
41
+
42
+
func TestGenerateTokenDistinct(t *testing.T) {
43
+
const n = 100
44
+
seen := make(map[string]struct{}, n)
45
+
for i := range n {
46
+
tok, err := GenerateToken()
47
+
if err != nil {
48
+
t.Fatalf("GenerateToken: %v", err)
49
+
}
50
+
if tok == "" {
51
+
t.Fatalf("GenerateToken returned an empty token on call %d", i)
52
+
}
53
+
if _, dup := seen[tok]; dup {
54
+
t.Fatalf("GenerateToken repeated a token after %d calls: %q", i, tok)
55
+
}
56
+
seen[tok] = struct{}{}
57
+
}
58
+
}
59
+
28
60
func TestAttachSessionRejectsSecondLiveSession(t *testing.T) {
29
61
l := discardLogger()
30
62
m := New(l, Config{ReconnectGrace: time.Minute})
+34
-8
spindle/mill/handler.go
+34
-8
spindle/mill/handler.go
···
2
2
3
3
import (
4
4
"net/http"
5
+
"strings"
5
6
6
7
"github.com/gorilla/websocket"
7
8
···
18
19
// secret in the Authorization header, checked before the upgrade so a bad token
19
20
// never opens a socket.
20
21
func (m *Mill) HandleExecutorConn(w http.ResponseWriter, r *http.Request) {
21
-
if m.cfg.SharedSecret != "" {
22
-
if r.Header.Get("Authorization") != "Bearer "+m.cfg.SharedSecret {
23
-
http.Error(w, "unauthorized", http.StatusUnauthorized)
24
-
return
25
-
}
22
+
name, ok := m.authenticate(r)
23
+
if !ok {
24
+
http.Error(w, "unauthorized", http.StatusUnauthorized)
25
+
return
26
26
}
27
27
28
28
conn, err := upgrader.Upgrade(w, r, nil)
···
52
52
return
53
53
}
54
54
55
-
sess := newSession(h.GetNodeId(), enc, m.l)
55
+
// identity comes from the authenticated token, never the client-supplied
56
+
// Hello node id, so a valid token can't impersonate another executor.
57
+
sess := newSession(name, enc, m.l)
58
+
sess.closeTransport = conn.Close
59
+
sess.labels = h.GetLabels()
56
60
resume, ok := m.attachSession(sess)
57
61
if !ok {
58
-
m.l.Warn("rejecting duplicate live executor session", "node", sess.nodeID)
62
+
m.l.Warn("rejecting duplicate live executor session", "node", name)
59
63
return
60
64
}
61
-
m.l.Info("executor connected", "node", sess.nodeID, "engines", h.GetEngines(), "arch", h.GetArch(), "resume", resume)
65
+
m.l.Info("executor connected", "node", sess.nodeID, "arch", h.GetArch(), "labels", h.GetLabels(), "resume", resume)
62
66
63
67
if err := sess.send(&millproto.Message{Resume: &millv1.Resume{AckOffset: resume}}); err != nil {
64
68
m.l.Error("fleet send resume failed", "err", err)
···
69
73
70
74
sess.readLoop(m, dec)
71
75
m.detachSession(sess)
76
+
}
77
+
78
+
// authenticate resolves the executor identity from the pre-shared token in the
79
+
// Authorization header before the upgrade, so a bad token never opens a socket.
80
+
// The token is looked up by hash; an unknown or missing token is rejected
81
+
// (fail closed).
82
+
func (m *Mill) authenticate(r *http.Request) (string, bool) {
83
+
const prefix = "Bearer "
84
+
h := r.Header.Get("Authorization")
85
+
if !strings.HasPrefix(h, prefix) {
86
+
return "", false
87
+
}
88
+
token := strings.TrimPrefix(h, prefix)
89
+
if token == "" || m.db == nil {
90
+
return "", false
91
+
}
92
+
name, ok, err := m.db.ResolveExecutorToken(HashToken(token))
93
+
if err != nil {
94
+
m.l.Error("executor token lookup failed", "err", err)
95
+
return "", false
96
+
}
97
+
return name, ok
72
98
}
+155
spindle/mill/integration_test.go
+155
spindle/mill/integration_test.go
···
34
34
bn := notifier.New()
35
35
mill := New(l, Config{LogDir: millDir, ReconnectGrace: time.Minute, BidTimeout: 2 * time.Second})
36
36
mill.Attach(bdb, &bn)
37
+
if err := bdb.AddExecutorToken("exec-1", HashToken("test-token"), nil); err != nil {
38
+
t.Fatalf("register executor token: %v", err)
39
+
}
37
40
38
41
srv := httptest.NewServer(http.HandlerFunc(mill.HandleExecutorConn))
39
42
defer srv.Close()
···
50
53
cfg.Server.Hostname = "exec-1"
51
54
cfg.Mill.URL = wsURL
52
55
cfg.Mill.Seats = 2
56
+
cfg.Mill.SharedSecret = "test-token"
53
57
54
58
engines := map[string]models.Engine{"dummy": dummy.New(l)}
55
59
exec := executor.New(cfg, engines, edb, &en, l)
···
82
86
if !waitForStatus(t, bdb, wid, "running") {
83
87
t.Fatal("mill never saw relayed running status")
84
88
}
89
+
}
90
+
91
+
func TestExecutorConfiguredLabelsAreStoredOnSession(t *testing.T) {
92
+
ctx, cancel := context.WithCancel(context.Background())
93
+
defer cancel()
94
+
l := slog.New(slog.NewTextHandler(io.Discard, nil))
95
+
96
+
millDir := t.TempDir()
97
+
bdb, err := db.Make(ctx, filepath.Join(millDir, "mill.db"))
98
+
if err != nil {
99
+
t.Fatalf("mill db: %v", err)
100
+
}
101
+
bn := notifier.New()
102
+
mill := New(l, Config{LogDir: millDir, ReconnectGrace: time.Minute, BidTimeout: 2 * time.Second})
103
+
mill.Attach(bdb, &bn)
104
+
if err := bdb.AddExecutorToken("exec-labels", HashToken("test-token"), nil); err != nil {
105
+
t.Fatalf("register executor token: %v", err)
106
+
}
107
+
108
+
srv := httptest.NewServer(http.HandlerFunc(mill.HandleExecutorConn))
109
+
defer srv.Close()
110
+
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http")
111
+
112
+
execDir := t.TempDir()
113
+
edb, err := db.Make(ctx, filepath.Join(execDir, "exec.db"))
114
+
if err != nil {
115
+
t.Fatalf("exec db: %v", err)
116
+
}
117
+
en := notifier.New()
118
+
cfg := &config.Config{}
119
+
cfg.Server.LogDir = execDir
120
+
cfg.Server.Hostname = "exec-labels"
121
+
cfg.Mill.URL = wsURL
122
+
cfg.Mill.Seats = 2
123
+
cfg.Mill.SharedSecret = "test-token"
124
+
cfg.Mill.Labels = []string{"linux", "arm64", "gpu"}
125
+
126
+
engines := map[string]models.Engine{"dummy": dummy.New(l)}
127
+
exec := executor.New(cfg, engines, edb, &en, l)
128
+
go exec.Connect(ctx)
129
+
130
+
if !waitForSessionLabels(t, mill, "exec-labels", []string{"linux", "arm64", "gpu"}) {
131
+
t.Fatal("mill session never stored executor labels from hello")
132
+
}
133
+
}
134
+
135
+
func TestEndToEndDummyJobUsesRequiredLabelsAcrossExecutors(t *testing.T) {
136
+
ctx, cancel := context.WithCancel(context.Background())
137
+
defer cancel()
138
+
l := slog.New(slog.NewTextHandler(io.Discard, nil))
139
+
140
+
millDir := t.TempDir()
141
+
bdb, err := db.Make(ctx, filepath.Join(millDir, "mill.db"))
142
+
if err != nil {
143
+
t.Fatalf("mill db: %v", err)
144
+
}
145
+
bn := notifier.New()
146
+
mill := New(l, Config{LogDir: millDir, ReconnectGrace: time.Minute, BidTimeout: 2 * time.Second})
147
+
mill.Attach(bdb, &bn)
148
+
if err := bdb.AddExecutorToken("exec-x86", HashToken("token-x86"), nil); err != nil {
149
+
t.Fatalf("register x86 executor token: %v", err)
150
+
}
151
+
if err := bdb.AddExecutorToken("exec-arm", HashToken("token-arm"), nil); err != nil {
152
+
t.Fatalf("register arm executor token: %v", err)
153
+
}
154
+
155
+
srv := httptest.NewServer(http.HandlerFunc(mill.HandleExecutorConn))
156
+
defer srv.Close()
157
+
wsURL := "ws" + strings.TrimPrefix(srv.URL, "http")
158
+
159
+
startExecutor := func(name, token string, labels []string) {
160
+
t.Helper()
161
+
execDir := t.TempDir()
162
+
edb, err := db.Make(ctx, filepath.Join(execDir, "exec.db"))
163
+
if err != nil {
164
+
t.Fatalf("%s exec db: %v", name, err)
165
+
}
166
+
en := notifier.New()
167
+
cfg := &config.Config{}
168
+
cfg.Server.LogDir = execDir
169
+
cfg.Server.Hostname = name
170
+
cfg.Mill.URL = wsURL
171
+
cfg.Mill.Seats = 1
172
+
cfg.Mill.SharedSecret = token
173
+
cfg.Mill.Labels = labels
174
+
175
+
engines := map[string]models.Engine{"dummy": dummy.New(l)}
176
+
exec := executor.New(cfg, engines, edb, &en, l)
177
+
go exec.Connect(ctx)
178
+
}
179
+
startExecutor("exec-x86", "token-x86", []string{"linux/amd64", "kvm"})
180
+
startExecutor("exec-arm", "token-arm", []string{"linux/arm64", "kvm"})
181
+
182
+
if !waitForSessionLabels(t, mill, "exec-x86", []string{"linux/amd64", "kvm"}) {
183
+
t.Fatal("x86 executor did not connect with labels")
184
+
}
185
+
if !waitForSessionLabels(t, mill, "exec-arm", []string{"linux/arm64", "kvm"}) {
186
+
t.Fatal("arm executor did not connect with labels")
187
+
}
188
+
189
+
be := NewEngine("dummy", mill)
190
+
twf := tangled.Pipeline_Workflow{
191
+
Name: "build-arm",
192
+
RunsOn: []string{"linux/arm64"},
193
+
Raw: "steps:\n - name: hello\n command: echo hi\n",
194
+
}
195
+
wf, err := be.InitWorkflow(twf, tangled.Pipeline{})
196
+
if err != nil {
197
+
t.Fatalf("InitWorkflow: %v", err)
198
+
}
199
+
wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "knot.test", Rkey: "rkey-arm"}, Name: "build-arm"}
200
+
201
+
placeCtx, placeCancel := context.WithTimeout(ctx, 10*time.Second)
202
+
defer placeCancel()
203
+
204
+
slot, err := mill.place(placeCtx, "dummy", wid, wf)
205
+
if err != nil {
206
+
t.Fatalf("place: %v", err)
207
+
}
208
+
defer slot.Release()
209
+
210
+
lease := wf.Data.(*millWorkflowState).Lease
211
+
if lease == nil {
212
+
t.Fatal("place did not attach lease to workflow state")
213
+
}
214
+
if lease.nodeID != "exec-arm" {
215
+
t.Fatalf("placed on %q, want exec-arm", lease.nodeID)
216
+
}
217
+
218
+
if err := mill.commitAndWait(placeCtx, wf, nil); err != nil {
219
+
t.Fatalf("commitAndWait: %v, want success", err)
220
+
}
221
+
}
222
+
223
+
func waitForSessionLabels(t *testing.T, m *Mill, nodeID string, want []string) bool {
224
+
t.Helper()
225
+
deadline := time.Now().Add(3 * time.Second)
226
+
for time.Now().Before(deadline) {
227
+
m.mu.Lock()
228
+
sess := m.sessions[nodeID]
229
+
var got []string
230
+
if sess != nil {
231
+
got = append([]string(nil), sess.labels...)
232
+
}
233
+
m.mu.Unlock()
234
+
if sameStringMultiset(got, want) {
235
+
return true
236
+
}
237
+
time.Sleep(10 * time.Millisecond)
238
+
}
239
+
return false
85
240
}
86
241
87
242
func waitForStatus(t *testing.T, d *db.DB, wid models.WorkflowId, want string) bool {
+25
-3
spindle/mill/mill.go
+25
-3
spindle/mill/mill.go
···
90
90
m.mu.Unlock()
91
91
}
92
92
93
+
// dev bootstrap: shared-secret identity, single-token fleet. production
94
+
// should register per-executor tokens instead.
95
+
func (m *Mill) SeedBootstrapToken() error {
96
+
if m.cfg.SharedSecret == "" || m.db == nil {
97
+
return nil
98
+
}
99
+
return m.db.UpsertExecutorToken(bootstrapTokenName, HashToken(m.cfg.SharedSecret))
100
+
}
101
+
93
102
func (m *Mill) nextLeaseID() string {
94
103
m.mu.Lock()
95
104
m.leaseSeq++
···
122
131
defer m.mu.Unlock()
123
132
124
133
if old := m.sessions[sess.nodeID]; old != nil {
125
-
if !old.disconnected {
134
+
if !old.disconnected && time.Since(old.lastSeen) <= m.cfg.ReconnectGrace {
126
135
// reject a second live session for the same identity so a valid token
127
136
// can't hijack an in-flight executor.
128
137
return 0, false
129
138
}
130
-
// the old session is in its reconnect grace window: adopt its leases.
131
139
if old.graceTimer != nil {
132
140
old.graceTimer.Stop()
133
141
}
134
142
old.close()
135
-
m.l.Info("executor reconnected", "node", sess.nodeID)
143
+
if old.disconnected {
144
+
m.l.Info("executor reconnected", "node", sess.nodeID)
145
+
} else {
146
+
m.l.Warn("replacing silent executor session", "node", sess.nodeID)
147
+
}
136
148
}
137
149
m.sessions[sess.nodeID] = sess
138
150
// wake commit retries waiting out this node's reconnect grace.
139
151
m.notifyChangeLocked()
140
152
return m.nodeOffset[sess.nodeID], true
153
+
}
154
+
155
+
func (m *Mill) touchSession(sess *millSession) bool {
156
+
m.mu.Lock()
157
+
defer m.mu.Unlock()
158
+
if m.sessions[sess.nodeID] != sess || sess.disconnected {
159
+
return false
160
+
}
161
+
sess.lastSeen = time.Now()
162
+
return true
141
163
}
142
164
143
165
func (m *Mill) detachSession(sess *millSession) {
+33
spindle/mill/mill_test.go
+33
spindle/mill/mill_test.go
···
495
495
t.Fatalf("bid winner = %+v, want fallback after silent incumbent times out", lease)
496
496
}
497
497
}
498
+
499
+
func TestAttachSessionReplacesSilentIncumbentButRejectsActiveDuplicate(t *testing.T) {
500
+
m := New(discardLogger(), Config{ReconnectGrace: time.Minute})
501
+
old := newSession("node-1", nopEncoder(), discardLogger())
502
+
transportClosed := make(chan struct{})
503
+
old.closeTransport = func() error {
504
+
close(transportClosed)
505
+
return nil
506
+
}
507
+
if _, ok := m.attachSession(old); !ok {
508
+
t.Fatal("first attach rejected")
509
+
}
510
+
m.mu.Lock()
511
+
old.lastSeen = time.Now().Add(-2 * m.cfg.ReconnectGrace)
512
+
m.mu.Unlock()
513
+
514
+
replacement := newSession("node-1", nopEncoder(), discardLogger())
515
+
if _, ok := m.attachSession(replacement); !ok {
516
+
t.Fatal("silent incumbent blocked authenticated replacement")
517
+
}
518
+
select {
519
+
case <-transportClosed:
520
+
default:
521
+
t.Fatal("replacing a silent incumbent did not close its transport")
522
+
}
523
+
m.mu.Lock()
524
+
replacement.lastSeen = time.Now().Add(-2 * m.cfg.ReconnectGrace)
525
+
m.mu.Unlock()
526
+
replacement.dispatch(m, &millproto.Message{NodeSnapshot: &millv1.NodeSnapshot{NodeId: "node-1"}})
527
+
if _, ok := m.attachSession(newSession("node-1", nopEncoder(), discardLogger())); ok {
528
+
t.Fatal("active replacement did not reject a duplicate session")
529
+
}
530
+
}
+21
-10
spindle/mill/session.go
+21
-10
spindle/mill/session.go
···
18
18
// it, so a single reader goroutine demuxes by message type and correlates
19
19
// request/response by lease id. We never hold a lock across a decode.
20
20
type millSession struct {
21
-
nodeID string
22
-
labels []string
23
-
enc messageEncoder
24
-
l *slog.Logger
21
+
nodeID string
22
+
labels []string
23
+
enc messageEncoder
24
+
l *slog.Logger
25
+
closeTransport func() error
25
26
26
27
// snapshot, disconnected and graceTimer are guarded by Mill.mu, not the
27
28
// session mutex below (the fleet ranks across sessions under its own lock).
28
29
snapshot *millv1.NodeSnapshot
29
30
disconnected bool
30
31
graceTimer *time.Timer
32
+
lastSeen time.Time
31
33
32
34
mu sync.Mutex
33
35
pending map[string]chan *millproto.Message // lease_id -> response waiter
···
42
44
43
45
func newSession(nodeID string, enc messageEncoder, l *slog.Logger) *millSession {
44
46
return &millSession{
45
-
nodeID: nodeID,
46
-
enc: enc,
47
-
l: l,
48
-
pending: make(map[string]chan *millproto.Message),
49
-
closed: make(chan struct{}),
47
+
nodeID: nodeID,
48
+
enc: enc,
49
+
l: l,
50
+
pending: make(map[string]chan *millproto.Message),
51
+
closed: make(chan struct{}),
52
+
lastSeen: time.Now(),
50
53
}
51
54
}
52
55
···
55
58
}
56
59
57
60
func (s *millSession) close() {
58
-
s.closeOnce.Do(func() { close(s.closed) })
61
+
s.closeOnce.Do(func() {
62
+
close(s.closed)
63
+
if s.closeTransport != nil {
64
+
_ = s.closeTransport()
65
+
}
66
+
})
59
67
}
60
68
61
69
// await registers a one-shot waiter for the next response addressed to leaseID,
···
124
132
}
125
133
126
134
func (s *millSession) dispatch(m *Mill, msg *millproto.Message) {
135
+
if !m.touchSession(s) {
136
+
return
137
+
}
127
138
switch {
128
139
case msg.GetNodeSnapshot() != nil:
129
140
m.onSnapshot(s, msg.GetNodeSnapshot())
+29
spindle/mill/token.go
+29
spindle/mill/token.go
···
1
+
package mill
2
+
3
+
import (
4
+
"crypto/rand"
5
+
"crypto/sha256"
6
+
"encoding/base64"
7
+
"encoding/hex"
8
+
)
9
+
10
+
// bootstrapTokenName is the identity the mill seeds cfg.SharedSecret under so a
11
+
// single-secret dev fleet works without minting. Production registers per-executor
12
+
// identities with `spindle mill executor add`.
13
+
const bootstrapTokenName = "dev-bootstrap"
14
+
15
+
func GenerateToken() (string, error) {
16
+
var b [32]byte
17
+
if _, err := rand.Read(b[:]); err != nil {
18
+
return "", err
19
+
}
20
+
return base64.RawURLEncoding.EncodeToString(b[:]), nil
21
+
}
22
+
23
+
// HashToken hashes a raw token for storage/lookup. The mill only ever persists
24
+
// and compares hashes, so a DB leak never exposes a usable token, and lookup by
25
+
// hash avoids a secret-dependent comparison.
26
+
func HashToken(token string) string {
27
+
sum := sha256.Sum256([]byte(token))
28
+
return hex.EncodeToString(sum[:])
29
+
}
-11
spindle/mill/executor/executor.go
-11
spindle/mill/executor/executor.go
···
152
152
153
153
hello := &millproto.Message{Hello: &millv1.Hello{
154
154
ProtocolVersion: millproto.ProtocolVersion,
155
-
NodeId: e.nodeID,
156
-
Engines: e.engineNames(),
157
155
Arch: runtime.GOARCH,
158
156
Labels: e.labels,
159
-
LastOffset: e.relay.lastOffset(),
160
157
}}
161
158
if err := enc.Encode(hello); err != nil {
162
159
return fmt.Errorf("send hello: %w", err)
···
525
522
e.draining = true
526
523
e.mu.Unlock()
527
524
e.pushSnapshot()
528
-
}
529
-
530
-
func (e *Executor) engineNames() []string {
531
-
names := make([]string, 0, len(e.engines))
532
-
for name := range e.engines {
533
-
names = append(names, name)
534
-
}
535
-
return names
536
525
}
537
526
538
527
func ttlDuration(secs uint32) time.Duration {
+2
-2
spindle/mill/executor/reserved.go
+2
-2
spindle/mill/executor/reserved.go
···
17
17
// This is the only change to the execution path on an executor.
18
18
type reservedEngine struct {
19
19
models.Engine
20
-
slot engine.WorkflowSlot
21
-
once sync.Once
20
+
slot engine.WorkflowSlot
21
+
once sync.Once
22
22
}
23
23
24
24
// newReservedEngine returns a wrapper around inner that hands back slot exactly
+6
-38
spindle/mill/proto/gen/mill.pb.go
+6
-38
spindle/mill/proto/gen/mill.pb.go
···
76
76
type Hello struct {
77
77
state protoimpl.MessageState `protogen:"open.v1"`
78
78
ProtocolVersion uint32 `protobuf:"varint,1,opt,name=protocol_version,json=protocolVersion,proto3" json:"protocol_version,omitempty"`
79
-
// stable across reconnects (e.g. hostname / DID); the mill keys sessions on
80
-
// this so a brief blip reattaches the same node rather than creating a new one.
81
-
NodeId string `protobuf:"bytes,2,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"`
82
-
// engine names this node can run ("microvm", "nixery").
83
-
Engines []string `protobuf:"bytes,3,rep,name=engines,proto3" json:"engines,omitempty"`
84
79
// GOARCH of the node, retained as an informational trait for logs.
85
-
Arch string `protobuf:"bytes,4,opt,name=arch,proto3" json:"arch,omitempty"`
86
-
// the highest relay offset the executor believes it has sent; a resume hint.
87
-
LastOffset uint64 `protobuf:"varint,5,opt,name=last_offset,json=lastOffset,proto3" json:"last_offset,omitempty"`
80
+
Arch string `protobuf:"bytes,2,opt,name=arch,proto3" json:"arch,omitempty"`
88
81
// opaque operator-defined labels used for runs_on matching.
89
-
Labels []string `protobuf:"bytes,6,rep,name=labels,proto3" json:"labels,omitempty"`
82
+
Labels []string `protobuf:"bytes,3,rep,name=labels,proto3" json:"labels,omitempty"`
90
83
unknownFields protoimpl.UnknownFields
91
84
sizeCache protoimpl.SizeCache
92
85
}
···
128
121
return 0
129
122
}
130
123
131
-
func (x *Hello) GetNodeId() string {
132
-
if x != nil {
133
-
return x.NodeId
134
-
}
135
-
return ""
136
-
}
137
-
138
-
func (x *Hello) GetEngines() []string {
139
-
if x != nil {
140
-
return x.Engines
141
-
}
142
-
return nil
143
-
}
144
-
145
124
func (x *Hello) GetArch() string {
146
125
if x != nil {
147
126
return x.Arch
148
127
}
149
128
return ""
150
-
}
151
-
152
-
func (x *Hello) GetLastOffset() uint64 {
153
-
if x != nil {
154
-
return x.LastOffset
155
-
}
156
-
return 0
157
129
}
158
130
159
131
func (x *Hello) GetLabels() []string {
···
1175
1147
1176
1148
const file_spindle_mill_v1_mill_proto_rawDesc = "" +
1177
1149
"\n" +
1178
-
"\x1aspindle/mill/v1/mill.proto\x12\x0fspindle.mill.v1\x1a\x1bbuf/validate/validate.proto\"\xbb\x01\n" +
1150
+
"\x1aspindle/mill/v1/mill.proto\x12\x0fspindle.mill.v1\x1a\x1bbuf/validate/validate.proto\"^\n" +
1179
1151
"\x05Hello\x12)\n" +
1180
-
"\x10protocol_version\x18\x01 \x01(\rR\x0fprotocolVersion\x12 \n" +
1181
-
"\anode_id\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\x06nodeId\x12\x18\n" +
1182
-
"\aengines\x18\x03 \x03(\tR\aengines\x12\x12\n" +
1183
-
"\x04arch\x18\x04 \x01(\tR\x04arch\x12\x1f\n" +
1184
-
"\vlast_offset\x18\x05 \x01(\x04R\n" +
1185
-
"lastOffset\x12\x16\n" +
1186
-
"\x06labels\x18\x06 \x03(\tR\x06labels\"'\n" +
1152
+
"\x10protocol_version\x18\x01 \x01(\rR\x0fprotocolVersion\x12\x12\n" +
1153
+
"\x04arch\x18\x02 \x01(\tR\x04arch\x12\x16\n" +
1154
+
"\x06labels\x18\x03 \x03(\tR\x06labels\"'\n" +
1187
1155
"\x06Resume\x12\x1d\n" +
1188
1156
"\n" +
1189
1157
"ack_offset\x18\x01 \x01(\x04R\tackOffset\"\xae\x01\n" +
+51
-52
shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs
+51
-52
shuttle/src/gen/spindle/agent/v1/spindle.agent.v1.rs
···
2
2
// This file is @generated by prost-build.
3
3
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
4
4
pub struct Hello {
5
-
#[prost(uint32, tag="1")]
5
+
#[prost(uint32, tag = "1")]
6
6
pub protocol_version: u32,
7
-
#[prost(string, tag="2")]
7
+
#[prost(string, tag = "2")]
8
8
pub agent_version: ::prost::alloc::string::String,
9
-
#[prost(string, tag="3")]
9
+
#[prost(string, tag = "3")]
10
10
pub boot_id: ::prost::alloc::string::String,
11
-
#[prost(string, tag="4")]
11
+
#[prost(string, tag = "4")]
12
12
pub nix_version: ::prost::alloc::string::String,
13
13
}
14
14
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
15
15
pub struct Init {
16
-
#[prost(string, tag="1")]
16
+
#[prost(string, tag = "1")]
17
17
pub job_id: ::prost::alloc::string::String,
18
-
#[prost(string, repeated, tag="2")]
18
+
#[prost(string, repeated, tag = "2")]
19
19
pub cache_trusted_public_keys: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
20
-
#[prost(uint32, tag="3")]
20
+
#[prost(uint32, tag = "3")]
21
21
pub cache_read_proxy_port: u32,
22
-
#[prost(uint32, tag="4")]
22
+
#[prost(uint32, tag = "4")]
23
23
pub cache_upload_proxy_port: u32,
24
-
#[prost(uint32, tag="5")]
24
+
#[prost(uint32, tag = "5")]
25
25
pub dns_proxy_port: u32,
26
26
}
27
27
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
28
28
pub struct ExecStart {
29
-
#[prost(string, repeated, tag="1")]
29
+
#[prost(string, repeated, tag = "1")]
30
30
pub argv: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
31
-
#[prost(string, repeated, tag="2")]
31
+
#[prost(string, repeated, tag = "2")]
32
32
pub env: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
33
-
#[prost(string, tag="3")]
33
+
#[prost(string, tag = "3")]
34
34
pub cwd: ::prost::alloc::string::String,
35
-
#[prost(string, tag="4")]
35
+
#[prost(string, tag = "4")]
36
36
pub user: ::prost::alloc::string::String,
37
-
#[prost(uint32, tag="5")]
37
+
#[prost(uint32, tag = "5")]
38
38
pub timeout_seconds: u32,
39
39
}
40
40
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
41
41
pub struct ExecStdout {
42
-
#[prost(string, tag="1")]
42
+
#[prost(string, tag = "1")]
43
43
pub data: ::prost::alloc::string::String,
44
44
}
45
45
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
46
46
pub struct ExecStderr {
47
-
#[prost(string, tag="1")]
47
+
#[prost(string, tag = "1")]
48
48
pub data: ::prost::alloc::string::String,
49
49
}
50
50
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
51
51
pub struct ExecExit {
52
-
#[prost(int32, tag="1")]
52
+
#[prost(int32, tag = "1")]
53
53
pub exit_code: i32,
54
-
#[prost(string, tag="2")]
54
+
#[prost(string, tag = "2")]
55
55
pub error: ::prost::alloc::string::String,
56
56
/// set when the guest killed the step on its own timeout timer, so the host
57
57
/// can classify it as a timeout rather than inferring failure from exit_code.
58
-
#[prost(bool, tag="3")]
58
+
#[prost(bool, tag = "3")]
59
59
pub timed_out: bool,
60
60
}
61
61
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
62
62
pub struct ActivateConfig {
63
-
#[prost(string, tag="1")]
63
+
#[prost(string, tag = "1")]
64
64
pub config_key: ::prost::alloc::string::String,
65
-
#[prost(string, tag="2")]
65
+
#[prost(string, tag = "2")]
66
66
pub base_config_hash: ::prost::alloc::string::String,
67
-
#[prost(string, tag="3")]
67
+
#[prost(string, tag = "3")]
68
68
pub user_config: ::prost::alloc::string::String,
69
-
#[prost(string, tag="4")]
69
+
#[prost(string, tag = "4")]
70
70
pub toplevel: ::prost::alloc::string::String,
71
-
#[prost(uint32, tag="5")]
71
+
#[prost(uint32, tag = "5")]
72
72
pub timeout_seconds: u32,
73
73
}
74
74
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
75
75
pub struct ActivateConfigResult {
76
-
#[prost(string, tag="1")]
76
+
#[prost(string, tag = "1")]
77
77
pub config_key: ::prost::alloc::string::String,
78
-
#[prost(string, tag="2")]
78
+
#[prost(string, tag = "2")]
79
79
pub toplevel: ::prost::alloc::string::String,
80
-
#[prost(string, tag="3")]
80
+
#[prost(string, tag = "3")]
81
81
pub error: ::prost::alloc::string::String,
82
82
}
83
83
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
84
84
pub struct BuiltPaths {
85
-
#[prost(string, repeated, tag="1")]
85
+
#[prost(string, repeated, tag = "1")]
86
86
pub paths: ::prost::alloc::vec::Vec<::prost::alloc::string::String>,
87
-
#[prost(string, tag="2")]
87
+
#[prost(string, tag = "2")]
88
88
pub reason: ::prost::alloc::string::String,
89
89
}
90
90
#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
91
91
pub struct CacheDrain {
92
-
#[prost(uint32, tag="1")]
92
+
#[prost(uint32, tag = "1")]
93
93
pub timeout_seconds: u32,
94
94
}
95
95
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
96
96
pub struct CacheDrainResult {
97
-
#[prost(string, tag="1")]
97
+
#[prost(string, tag = "1")]
98
98
pub error: ::prost::alloc::string::String,
99
-
#[prost(uint32, tag="2")]
99
+
#[prost(uint32, tag = "2")]
100
100
pub cache_queued: u32,
101
-
#[prost(uint32, tag="3")]
101
+
#[prost(uint32, tag = "3")]
102
102
pub cache_active: u32,
103
-
#[prost(uint32, tag="4")]
103
+
#[prost(uint32, tag = "4")]
104
104
pub cache_uploaded: u32,
105
-
#[prost(uint32, tag="5")]
105
+
#[prost(uint32, tag = "5")]
106
106
pub cache_failed: u32,
107
107
}
108
108
#[derive(Clone, Copy, PartialEq, Eq, Hash, ::prost::Message)]
109
-
pub struct Poweroff {
110
-
}
109
+
pub struct Poweroff {}
111
110
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
112
111
pub struct PoweroffResult {
113
-
#[prost(string, tag="1")]
112
+
#[prost(string, tag = "1")]
114
113
pub error: ::prost::alloc::string::String,
115
114
}
116
115
#[derive(Clone, PartialEq, Eq, Hash, ::prost::Message)]
117
116
pub struct Message {
118
-
#[prost(string, tag="1")]
117
+
#[prost(string, tag = "1")]
119
118
pub id: ::prost::alloc::string::String,
120
-
#[prost(message, optional, tag="2")]
119
+
#[prost(message, optional, tag = "2")]
121
120
pub hello: ::core::option::Option<Hello>,
122
-
#[prost(message, optional, tag="3")]
121
+
#[prost(message, optional, tag = "3")]
123
122
pub init: ::core::option::Option<Init>,
124
-
#[prost(message, optional, tag="4")]
123
+
#[prost(message, optional, tag = "4")]
125
124
pub exec_start: ::core::option::Option<ExecStart>,
126
-
#[prost(message, optional, tag="5")]
125
+
#[prost(message, optional, tag = "5")]
127
126
pub exec_stdout: ::core::option::Option<ExecStdout>,
128
-
#[prost(message, optional, tag="6")]
127
+
#[prost(message, optional, tag = "6")]
129
128
pub exec_stderr: ::core::option::Option<ExecStderr>,
130
-
#[prost(message, optional, tag="7")]
129
+
#[prost(message, optional, tag = "7")]
131
130
pub exec_exit: ::core::option::Option<ExecExit>,
132
-
#[prost(message, optional, tag="8")]
131
+
#[prost(message, optional, tag = "8")]
133
132
pub activate_config: ::core::option::Option<ActivateConfig>,
134
-
#[prost(message, optional, tag="9")]
133
+
#[prost(message, optional, tag = "9")]
135
134
pub activate_config_result: ::core::option::Option<ActivateConfigResult>,
136
-
#[prost(message, optional, tag="10")]
135
+
#[prost(message, optional, tag = "10")]
137
136
pub built_paths: ::core::option::Option<BuiltPaths>,
138
-
#[prost(message, optional, tag="11")]
137
+
#[prost(message, optional, tag = "11")]
139
138
pub cache_drain: ::core::option::Option<CacheDrain>,
140
-
#[prost(message, optional, tag="12")]
139
+
#[prost(message, optional, tag = "12")]
141
140
pub cache_drain_result: ::core::option::Option<CacheDrainResult>,
142
-
#[prost(message, optional, tag="13")]
141
+
#[prost(message, optional, tag = "13")]
143
142
pub poweroff: ::core::option::Option<Poweroff>,
144
-
#[prost(message, optional, tag="14")]
143
+
#[prost(message, optional, tag = "14")]
145
144
pub poweroff_result: ::core::option::Option<PoweroffResult>,
146
145
}
147
146
// @@protoc_insertion_point(module)
+2
-9
spindle/mill/proto/spindle/mill/v1/mill.proto
+2
-9
spindle/mill/proto/spindle/mill/v1/mill.proto
···
10
10
// carries the static traits of the node plus a resume hint.
11
11
message Hello {
12
12
uint32 protocol_version = 1;
13
-
// stable across reconnects (e.g. hostname / DID); the mill keys sessions on
14
-
// this so a brief blip reattaches the same node rather than creating a new one.
15
-
string node_id = 2 [(buf.validate.field).string.min_len = 1];
16
-
// engine names this node can run ("microvm", "nixery").
17
-
repeated string engines = 3;
18
13
// GOARCH of the node, retained as an informational trait for logs.
19
-
string arch = 4;
20
-
// the highest relay offset the executor believes it has sent; a resume hint.
21
-
uint64 last_offset = 5;
14
+
string arch = 2;
22
15
// opaque operator-defined labels used for runs_on matching.
23
-
repeated string labels = 6;
16
+
repeated string labels = 3;
24
17
}
25
18
26
19
// Resume is the mill's reply to Hello. The executor replays every buffered