Monorepo for Tangled
spindle/mill: add executor wire protocol
+1721
+4
buf.gen.yaml
+4
buf.gen.yaml
+1
buf.yaml
+1
buf.yaml
+1317
spindle/mill/proto/gen/mill.pb.go
+1317
spindle/mill/proto/gen/mill.pb.go
···
1
+
// Code generated by protoc-gen-go. DO NOT EDIT.
2
+
// versions:
3
+
// protoc-gen-go v1.36.11
4
+
// protoc (unknown)
5
+
// source: spindle/mill/v1/mill.proto
6
+
7
+
package millv1
8
+
9
+
import (
10
+
_ "buf.build/gen/go/bufbuild/protovalidate/protocolbuffers/go/buf/validate"
11
+
protoreflect "google.golang.org/protobuf/reflect/protoreflect"
12
+
protoimpl "google.golang.org/protobuf/runtime/protoimpl"
13
+
reflect "reflect"
14
+
sync "sync"
15
+
unsafe "unsafe"
16
+
)
17
+
18
+
const (
19
+
// Verify that this generated code is sufficiently up-to-date.
20
+
_ = protoimpl.EnforceVersion(20 - protoimpl.MinVersion)
21
+
// Verify that runtime/protoimpl is sufficiently up-to-date.
22
+
_ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20)
23
+
)
24
+
25
+
// Hello is the first frame an executor sends after dialing the mill. It
26
+
// carries the static traits of the node plus a resume hint.
27
+
type Hello struct {
28
+
state protoimpl.MessageState `protogen:"open.v1"`
29
+
ProtocolVersion uint32 `protobuf:"varint,1,opt,name=protocol_version,json=protocolVersion,proto3" json:"protocol_version,omitempty"`
30
+
// stable across reconnects (e.g. hostname / DID); the mill keys sessions on
31
+
// this so a brief blip reattaches the same node rather than creating a new one.
32
+
NodeId string `protobuf:"bytes,2,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"`
33
+
// engine names this node can run ("microvm", "nixery").
34
+
Engines []string `protobuf:"bytes,3,rep,name=engines,proto3" json:"engines,omitempty"`
35
+
// GOARCH of the node, so the mill won't place arch-incompatible jobs.
36
+
Arch string `protobuf:"bytes,4,opt,name=arch,proto3" json:"arch,omitempty"`
37
+
// the highest relay offset the executor believes it has sent; a resume hint.
38
+
LastOffset uint64 `protobuf:"varint,5,opt,name=last_offset,json=lastOffset,proto3" json:"last_offset,omitempty"`
39
+
unknownFields protoimpl.UnknownFields
40
+
sizeCache protoimpl.SizeCache
41
+
}
42
+
43
+
func (x *Hello) Reset() {
44
+
*x = Hello{}
45
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[0]
46
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
47
+
ms.StoreMessageInfo(mi)
48
+
}
49
+
50
+
func (x *Hello) String() string {
51
+
return protoimpl.X.MessageStringOf(x)
52
+
}
53
+
54
+
func (*Hello) ProtoMessage() {}
55
+
56
+
func (x *Hello) ProtoReflect() protoreflect.Message {
57
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[0]
58
+
if x != nil {
59
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
60
+
if ms.LoadMessageInfo() == nil {
61
+
ms.StoreMessageInfo(mi)
62
+
}
63
+
return ms
64
+
}
65
+
return mi.MessageOf(x)
66
+
}
67
+
68
+
// Deprecated: Use Hello.ProtoReflect.Descriptor instead.
69
+
func (*Hello) Descriptor() ([]byte, []int) {
70
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{0}
71
+
}
72
+
73
+
func (x *Hello) GetProtocolVersion() uint32 {
74
+
if x != nil {
75
+
return x.ProtocolVersion
76
+
}
77
+
return 0
78
+
}
79
+
80
+
func (x *Hello) GetNodeId() string {
81
+
if x != nil {
82
+
return x.NodeId
83
+
}
84
+
return ""
85
+
}
86
+
87
+
func (x *Hello) GetEngines() []string {
88
+
if x != nil {
89
+
return x.Engines
90
+
}
91
+
return nil
92
+
}
93
+
94
+
func (x *Hello) GetArch() string {
95
+
if x != nil {
96
+
return x.Arch
97
+
}
98
+
return ""
99
+
}
100
+
101
+
func (x *Hello) GetLastOffset() uint64 {
102
+
if x != nil {
103
+
return x.LastOffset
104
+
}
105
+
return 0
106
+
}
107
+
108
+
// Resume is the mill's reply to Hello. The executor replays every buffered
109
+
// relay entry with offset strictly greater than ack_offset before sending new
110
+
// ones.
111
+
type Resume struct {
112
+
state protoimpl.MessageState `protogen:"open.v1"`
113
+
AckOffset uint64 `protobuf:"varint,1,opt,name=ack_offset,json=ackOffset,proto3" json:"ack_offset,omitempty"`
114
+
unknownFields protoimpl.UnknownFields
115
+
sizeCache protoimpl.SizeCache
116
+
}
117
+
118
+
func (x *Resume) Reset() {
119
+
*x = Resume{}
120
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[1]
121
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
122
+
ms.StoreMessageInfo(mi)
123
+
}
124
+
125
+
func (x *Resume) String() string {
126
+
return protoimpl.X.MessageStringOf(x)
127
+
}
128
+
129
+
func (*Resume) ProtoMessage() {}
130
+
131
+
func (x *Resume) ProtoReflect() protoreflect.Message {
132
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[1]
133
+
if x != nil {
134
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
135
+
if ms.LoadMessageInfo() == nil {
136
+
ms.StoreMessageInfo(mi)
137
+
}
138
+
return ms
139
+
}
140
+
return mi.MessageOf(x)
141
+
}
142
+
143
+
// Deprecated: Use Resume.ProtoReflect.Descriptor instead.
144
+
func (*Resume) Descriptor() ([]byte, []int) {
145
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{1}
146
+
}
147
+
148
+
func (x *Resume) GetAckOffset() uint64 {
149
+
if x != nil {
150
+
return x.AckOffset
151
+
}
152
+
return 0
153
+
}
154
+
155
+
// EngineSnapshot is the changing per-engine state on a node. free_seats is a
156
+
// coarse "can you take more" hint; 0 also means draining (the mill treats a
157
+
// draining node as a full one until it leaves). The resource fields are coarse
158
+
// budget headroom for ranking only; the executor makes the real yes/no call in
159
+
// ReserveResult.
160
+
type EngineSnapshot struct {
161
+
state protoimpl.MessageState `protogen:"open.v1"`
162
+
FreeSeats uint32 `protobuf:"varint,1,opt,name=free_seats,json=freeSeats,proto3" json:"free_seats,omitempty"`
163
+
FreeMemoryMib int64 `protobuf:"varint,2,opt,name=free_memory_mib,json=freeMemoryMib,proto3" json:"free_memory_mib,omitempty"`
164
+
FreeVcpus int64 `protobuf:"varint,3,opt,name=free_vcpus,json=freeVcpus,proto3" json:"free_vcpus,omitempty"`
165
+
FreeDiskMib int64 `protobuf:"varint,4,opt,name=free_disk_mib,json=freeDiskMib,proto3" json:"free_disk_mib,omitempty"`
166
+
unknownFields protoimpl.UnknownFields
167
+
sizeCache protoimpl.SizeCache
168
+
}
169
+
170
+
func (x *EngineSnapshot) Reset() {
171
+
*x = EngineSnapshot{}
172
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[2]
173
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
174
+
ms.StoreMessageInfo(mi)
175
+
}
176
+
177
+
func (x *EngineSnapshot) String() string {
178
+
return protoimpl.X.MessageStringOf(x)
179
+
}
180
+
181
+
func (*EngineSnapshot) ProtoMessage() {}
182
+
183
+
func (x *EngineSnapshot) ProtoReflect() protoreflect.Message {
184
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[2]
185
+
if x != nil {
186
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
187
+
if ms.LoadMessageInfo() == nil {
188
+
ms.StoreMessageInfo(mi)
189
+
}
190
+
return ms
191
+
}
192
+
return mi.MessageOf(x)
193
+
}
194
+
195
+
// Deprecated: Use EngineSnapshot.ProtoReflect.Descriptor instead.
196
+
func (*EngineSnapshot) Descriptor() ([]byte, []int) {
197
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{2}
198
+
}
199
+
200
+
func (x *EngineSnapshot) GetFreeSeats() uint32 {
201
+
if x != nil {
202
+
return x.FreeSeats
203
+
}
204
+
return 0
205
+
}
206
+
207
+
func (x *EngineSnapshot) GetFreeMemoryMib() int64 {
208
+
if x != nil {
209
+
return x.FreeMemoryMib
210
+
}
211
+
return 0
212
+
}
213
+
214
+
func (x *EngineSnapshot) GetFreeVcpus() int64 {
215
+
if x != nil {
216
+
return x.FreeVcpus
217
+
}
218
+
return 0
219
+
}
220
+
221
+
func (x *EngineSnapshot) GetFreeDiskMib() int64 {
222
+
if x != nil {
223
+
return x.FreeDiskMib
224
+
}
225
+
return 0
226
+
}
227
+
228
+
// NodeSnapshot is pushed on connect, periodically, and right after any state
229
+
// change (reserve, commit, terminal).
230
+
type NodeSnapshot struct {
231
+
state protoimpl.MessageState `protogen:"open.v1"`
232
+
NodeId string `protobuf:"bytes,1,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"`
233
+
Seq uint64 `protobuf:"varint,2,opt,name=seq,proto3" json:"seq,omitempty"`
234
+
Engines map[string]*EngineSnapshot `protobuf:"bytes,3,rep,name=engines,proto3" json:"engines,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"bytes,2,opt,name=value"`
235
+
unknownFields protoimpl.UnknownFields
236
+
sizeCache protoimpl.SizeCache
237
+
}
238
+
239
+
func (x *NodeSnapshot) Reset() {
240
+
*x = NodeSnapshot{}
241
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[3]
242
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
243
+
ms.StoreMessageInfo(mi)
244
+
}
245
+
246
+
func (x *NodeSnapshot) String() string {
247
+
return protoimpl.X.MessageStringOf(x)
248
+
}
249
+
250
+
func (*NodeSnapshot) ProtoMessage() {}
251
+
252
+
func (x *NodeSnapshot) ProtoReflect() protoreflect.Message {
253
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[3]
254
+
if x != nil {
255
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
256
+
if ms.LoadMessageInfo() == nil {
257
+
ms.StoreMessageInfo(mi)
258
+
}
259
+
return ms
260
+
}
261
+
return mi.MessageOf(x)
262
+
}
263
+
264
+
// Deprecated: Use NodeSnapshot.ProtoReflect.Descriptor instead.
265
+
func (*NodeSnapshot) Descriptor() ([]byte, []int) {
266
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{3}
267
+
}
268
+
269
+
func (x *NodeSnapshot) GetNodeId() string {
270
+
if x != nil {
271
+
return x.NodeId
272
+
}
273
+
return ""
274
+
}
275
+
276
+
func (x *NodeSnapshot) GetSeq() uint64 {
277
+
if x != nil {
278
+
return x.Seq
279
+
}
280
+
return 0
281
+
}
282
+
283
+
func (x *NodeSnapshot) GetEngines() map[string]*EngineSnapshot {
284
+
if x != nil {
285
+
return x.Engines
286
+
}
287
+
return nil
288
+
}
289
+
290
+
// ReserveSeat asks an executor to hold a seat for a job. Zero secrets ride this
291
+
// message; the raw pipeline/workflow are carried as JSON since processPipeline
292
+
// already round-trips them through JSON.
293
+
type ReserveSeat struct {
294
+
state protoimpl.MessageState `protogen:"open.v1"`
295
+
LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"`
296
+
TargetEngine string `protobuf:"bytes,2,opt,name=target_engine,json=targetEngine,proto3" json:"target_engine,omitempty"`
297
+
RawPipelineJson string `protobuf:"bytes,3,opt,name=raw_pipeline_json,json=rawPipelineJson,proto3" json:"raw_pipeline_json,omitempty"`
298
+
RawWorkflowJson string `protobuf:"bytes,4,opt,name=raw_workflow_json,json=rawWorkflowJson,proto3" json:"raw_workflow_json,omitempty"`
299
+
// the pipeline id (knot + rkey); the executor reconstructs the exact
300
+
// WorkflowId so its relayed status rows and log path match what the mill
301
+
// authored for "pending".
302
+
Knot string `protobuf:"bytes,5,opt,name=knot,proto3" json:"knot,omitempty"`
303
+
Rkey string `protobuf:"bytes,6,opt,name=rkey,proto3" json:"rkey,omitempty"`
304
+
TtlSeconds uint32 `protobuf:"varint,7,opt,name=ttl_seconds,json=ttlSeconds,proto3" json:"ttl_seconds,omitempty"`
305
+
unknownFields protoimpl.UnknownFields
306
+
sizeCache protoimpl.SizeCache
307
+
}
308
+
309
+
func (x *ReserveSeat) Reset() {
310
+
*x = ReserveSeat{}
311
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[4]
312
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
313
+
ms.StoreMessageInfo(mi)
314
+
}
315
+
316
+
func (x *ReserveSeat) String() string {
317
+
return protoimpl.X.MessageStringOf(x)
318
+
}
319
+
320
+
func (*ReserveSeat) ProtoMessage() {}
321
+
322
+
func (x *ReserveSeat) ProtoReflect() protoreflect.Message {
323
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[4]
324
+
if x != nil {
325
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
326
+
if ms.LoadMessageInfo() == nil {
327
+
ms.StoreMessageInfo(mi)
328
+
}
329
+
return ms
330
+
}
331
+
return mi.MessageOf(x)
332
+
}
333
+
334
+
// Deprecated: Use ReserveSeat.ProtoReflect.Descriptor instead.
335
+
func (*ReserveSeat) Descriptor() ([]byte, []int) {
336
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{4}
337
+
}
338
+
339
+
func (x *ReserveSeat) GetLeaseId() string {
340
+
if x != nil {
341
+
return x.LeaseId
342
+
}
343
+
return ""
344
+
}
345
+
346
+
func (x *ReserveSeat) GetTargetEngine() string {
347
+
if x != nil {
348
+
return x.TargetEngine
349
+
}
350
+
return ""
351
+
}
352
+
353
+
func (x *ReserveSeat) GetRawPipelineJson() string {
354
+
if x != nil {
355
+
return x.RawPipelineJson
356
+
}
357
+
return ""
358
+
}
359
+
360
+
func (x *ReserveSeat) GetRawWorkflowJson() string {
361
+
if x != nil {
362
+
return x.RawWorkflowJson
363
+
}
364
+
return ""
365
+
}
366
+
367
+
func (x *ReserveSeat) GetKnot() string {
368
+
if x != nil {
369
+
return x.Knot
370
+
}
371
+
return ""
372
+
}
373
+
374
+
func (x *ReserveSeat) GetRkey() string {
375
+
if x != nil {
376
+
return x.Rkey
377
+
}
378
+
return ""
379
+
}
380
+
381
+
func (x *ReserveSeat) GetTtlSeconds() uint32 {
382
+
if x != nil {
383
+
return x.TtlSeconds
384
+
}
385
+
return 0
386
+
}
387
+
388
+
// ReserveResult is the executor's accept/reject for a ReserveSeat.
389
+
type ReserveResult struct {
390
+
state protoimpl.MessageState `protogen:"open.v1"`
391
+
LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"`
392
+
Accepted bool `protobuf:"varint,2,opt,name=accepted,proto3" json:"accepted,omitempty"`
393
+
RejectReason string `protobuf:"bytes,3,opt,name=reject_reason,json=rejectReason,proto3" json:"reject_reason,omitempty"`
394
+
// optional bid score the mill ranks accepted leases by (higher is better).
395
+
Score float64 `protobuf:"fixed64,4,opt,name=score,proto3" json:"score,omitempty"`
396
+
unknownFields protoimpl.UnknownFields
397
+
sizeCache protoimpl.SizeCache
398
+
}
399
+
400
+
func (x *ReserveResult) Reset() {
401
+
*x = ReserveResult{}
402
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[5]
403
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
404
+
ms.StoreMessageInfo(mi)
405
+
}
406
+
407
+
func (x *ReserveResult) String() string {
408
+
return protoimpl.X.MessageStringOf(x)
409
+
}
410
+
411
+
func (*ReserveResult) ProtoMessage() {}
412
+
413
+
func (x *ReserveResult) ProtoReflect() protoreflect.Message {
414
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[5]
415
+
if x != nil {
416
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
417
+
if ms.LoadMessageInfo() == nil {
418
+
ms.StoreMessageInfo(mi)
419
+
}
420
+
return ms
421
+
}
422
+
return mi.MessageOf(x)
423
+
}
424
+
425
+
// Deprecated: Use ReserveResult.ProtoReflect.Descriptor instead.
426
+
func (*ReserveResult) Descriptor() ([]byte, []int) {
427
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{5}
428
+
}
429
+
430
+
func (x *ReserveResult) GetLeaseId() string {
431
+
if x != nil {
432
+
return x.LeaseId
433
+
}
434
+
return ""
435
+
}
436
+
437
+
func (x *ReserveResult) GetAccepted() bool {
438
+
if x != nil {
439
+
return x.Accepted
440
+
}
441
+
return false
442
+
}
443
+
444
+
func (x *ReserveResult) GetRejectReason() string {
445
+
if x != nil {
446
+
return x.RejectReason
447
+
}
448
+
return ""
449
+
}
450
+
451
+
func (x *ReserveResult) GetScore() float64 {
452
+
if x != nil {
453
+
return x.Score
454
+
}
455
+
return 0
456
+
}
457
+
458
+
// Secret is a single unlocked secret, sent only inside CommitLease.
459
+
type Secret struct {
460
+
state protoimpl.MessageState `protogen:"open.v1"`
461
+
Key string `protobuf:"bytes,1,opt,name=key,proto3" json:"key,omitempty"`
462
+
Value string `protobuf:"bytes,2,opt,name=value,proto3" json:"value,omitempty"`
463
+
unknownFields protoimpl.UnknownFields
464
+
sizeCache protoimpl.SizeCache
465
+
}
466
+
467
+
func (x *Secret) Reset() {
468
+
*x = Secret{}
469
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[6]
470
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
471
+
ms.StoreMessageInfo(mi)
472
+
}
473
+
474
+
func (x *Secret) String() string {
475
+
return protoimpl.X.MessageStringOf(x)
476
+
}
477
+
478
+
func (*Secret) ProtoMessage() {}
479
+
480
+
func (x *Secret) ProtoReflect() protoreflect.Message {
481
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[6]
482
+
if x != nil {
483
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
484
+
if ms.LoadMessageInfo() == nil {
485
+
ms.StoreMessageInfo(mi)
486
+
}
487
+
return ms
488
+
}
489
+
return mi.MessageOf(x)
490
+
}
491
+
492
+
// Deprecated: Use Secret.ProtoReflect.Descriptor instead.
493
+
func (*Secret) Descriptor() ([]byte, []int) {
494
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{6}
495
+
}
496
+
497
+
func (x *Secret) GetKey() string {
498
+
if x != nil {
499
+
return x.Key
500
+
}
501
+
return ""
502
+
}
503
+
504
+
func (x *Secret) GetValue() string {
505
+
if x != nil {
506
+
return x.Value
507
+
}
508
+
return ""
509
+
}
510
+
511
+
// CommitLease promotes a reservation to a running job and hands over the
512
+
// secrets. The executor then runs the real engine with the slot it already
513
+
// holds.
514
+
type CommitLease struct {
515
+
state protoimpl.MessageState `protogen:"open.v1"`
516
+
LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"`
517
+
Secrets []*Secret `protobuf:"bytes,2,rep,name=secrets,proto3" json:"secrets,omitempty"`
518
+
unknownFields protoimpl.UnknownFields
519
+
sizeCache protoimpl.SizeCache
520
+
}
521
+
522
+
func (x *CommitLease) Reset() {
523
+
*x = CommitLease{}
524
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[7]
525
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
526
+
ms.StoreMessageInfo(mi)
527
+
}
528
+
529
+
func (x *CommitLease) String() string {
530
+
return protoimpl.X.MessageStringOf(x)
531
+
}
532
+
533
+
func (*CommitLease) ProtoMessage() {}
534
+
535
+
func (x *CommitLease) ProtoReflect() protoreflect.Message {
536
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[7]
537
+
if x != nil {
538
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
539
+
if ms.LoadMessageInfo() == nil {
540
+
ms.StoreMessageInfo(mi)
541
+
}
542
+
return ms
543
+
}
544
+
return mi.MessageOf(x)
545
+
}
546
+
547
+
// Deprecated: Use CommitLease.ProtoReflect.Descriptor instead.
548
+
func (*CommitLease) Descriptor() ([]byte, []int) {
549
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{7}
550
+
}
551
+
552
+
func (x *CommitLease) GetLeaseId() string {
553
+
if x != nil {
554
+
return x.LeaseId
555
+
}
556
+
return ""
557
+
}
558
+
559
+
func (x *CommitLease) GetSecrets() []*Secret {
560
+
if x != nil {
561
+
return x.Secrets
562
+
}
563
+
return nil
564
+
}
565
+
566
+
// Committed acks a CommitLease.
567
+
type Committed struct {
568
+
state protoimpl.MessageState `protogen:"open.v1"`
569
+
LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"`
570
+
unknownFields protoimpl.UnknownFields
571
+
sizeCache protoimpl.SizeCache
572
+
}
573
+
574
+
func (x *Committed) Reset() {
575
+
*x = Committed{}
576
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[8]
577
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
578
+
ms.StoreMessageInfo(mi)
579
+
}
580
+
581
+
func (x *Committed) String() string {
582
+
return protoimpl.X.MessageStringOf(x)
583
+
}
584
+
585
+
func (*Committed) ProtoMessage() {}
586
+
587
+
func (x *Committed) ProtoReflect() protoreflect.Message {
588
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[8]
589
+
if x != nil {
590
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
591
+
if ms.LoadMessageInfo() == nil {
592
+
ms.StoreMessageInfo(mi)
593
+
}
594
+
return ms
595
+
}
596
+
return mi.MessageOf(x)
597
+
}
598
+
599
+
// Deprecated: Use Committed.ProtoReflect.Descriptor instead.
600
+
func (*Committed) Descriptor() ([]byte, []int) {
601
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{8}
602
+
}
603
+
604
+
func (x *Committed) GetLeaseId() string {
605
+
if x != nil {
606
+
return x.LeaseId
607
+
}
608
+
return ""
609
+
}
610
+
611
+
// ReleaseLease tells the executor to drop a reservation it never committed (the
612
+
// mill picked another node, or is cleaning up).
613
+
type ReleaseLease struct {
614
+
state protoimpl.MessageState `protogen:"open.v1"`
615
+
LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"`
616
+
unknownFields protoimpl.UnknownFields
617
+
sizeCache protoimpl.SizeCache
618
+
}
619
+
620
+
func (x *ReleaseLease) Reset() {
621
+
*x = ReleaseLease{}
622
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[9]
623
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
624
+
ms.StoreMessageInfo(mi)
625
+
}
626
+
627
+
func (x *ReleaseLease) String() string {
628
+
return protoimpl.X.MessageStringOf(x)
629
+
}
630
+
631
+
func (*ReleaseLease) ProtoMessage() {}
632
+
633
+
func (x *ReleaseLease) ProtoReflect() protoreflect.Message {
634
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[9]
635
+
if x != nil {
636
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
637
+
if ms.LoadMessageInfo() == nil {
638
+
ms.StoreMessageInfo(mi)
639
+
}
640
+
return ms
641
+
}
642
+
return mi.MessageOf(x)
643
+
}
644
+
645
+
// Deprecated: Use ReleaseLease.ProtoReflect.Descriptor instead.
646
+
func (*ReleaseLease) Descriptor() ([]byte, []int) {
647
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{9}
648
+
}
649
+
650
+
func (x *ReleaseLease) GetLeaseId() string {
651
+
if x != nil {
652
+
return x.LeaseId
653
+
}
654
+
return ""
655
+
}
656
+
657
+
// CancelAttempt cancels a running attempt (user cancel / DestroyWorkflow).
658
+
type CancelAttempt struct {
659
+
state protoimpl.MessageState `protogen:"open.v1"`
660
+
LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"`
661
+
Reason string `protobuf:"bytes,2,opt,name=reason,proto3" json:"reason,omitempty"`
662
+
unknownFields protoimpl.UnknownFields
663
+
sizeCache protoimpl.SizeCache
664
+
}
665
+
666
+
func (x *CancelAttempt) Reset() {
667
+
*x = CancelAttempt{}
668
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[10]
669
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
670
+
ms.StoreMessageInfo(mi)
671
+
}
672
+
673
+
func (x *CancelAttempt) String() string {
674
+
return protoimpl.X.MessageStringOf(x)
675
+
}
676
+
677
+
func (*CancelAttempt) ProtoMessage() {}
678
+
679
+
func (x *CancelAttempt) ProtoReflect() protoreflect.Message {
680
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[10]
681
+
if x != nil {
682
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
683
+
if ms.LoadMessageInfo() == nil {
684
+
ms.StoreMessageInfo(mi)
685
+
}
686
+
return ms
687
+
}
688
+
return mi.MessageOf(x)
689
+
}
690
+
691
+
// Deprecated: Use CancelAttempt.ProtoReflect.Descriptor instead.
692
+
func (*CancelAttempt) Descriptor() ([]byte, []int) {
693
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{10}
694
+
}
695
+
696
+
func (x *CancelAttempt) GetLeaseId() string {
697
+
if x != nil {
698
+
return x.LeaseId
699
+
}
700
+
return ""
701
+
}
702
+
703
+
func (x *CancelAttempt) GetReason() string {
704
+
if x != nil {
705
+
return x.Reason
706
+
}
707
+
return ""
708
+
}
709
+
710
+
// StatusEvent relays a non-terminal status (i.e. running) the executor wrote to
711
+
// its own eventstream. Terminals are authored by the mill from the attempt
712
+
// result, so they are never relayed here. Carries a per-session monotonic
713
+
// offset for gap-free replay across reconnects.
714
+
type StatusEvent struct {
715
+
state protoimpl.MessageState `protogen:"open.v1"`
716
+
Offset uint64 `protobuf:"varint,1,opt,name=offset,proto3" json:"offset,omitempty"`
717
+
LeaseId string `protobuf:"bytes,2,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"`
718
+
Status string `protobuf:"bytes,3,opt,name=status,proto3" json:"status,omitempty"`
719
+
Error string `protobuf:"bytes,4,opt,name=error,proto3" json:"error,omitempty"`
720
+
ExitCode int64 `protobuf:"varint,5,opt,name=exit_code,json=exitCode,proto3" json:"exit_code,omitempty"`
721
+
Workflow string `protobuf:"bytes,6,opt,name=workflow,proto3" json:"workflow,omitempty"`
722
+
PipelineAturi string `protobuf:"bytes,7,opt,name=pipeline_aturi,json=pipelineAturi,proto3" json:"pipeline_aturi,omitempty"`
723
+
unknownFields protoimpl.UnknownFields
724
+
sizeCache protoimpl.SizeCache
725
+
}
726
+
727
+
func (x *StatusEvent) Reset() {
728
+
*x = StatusEvent{}
729
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[11]
730
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
731
+
ms.StoreMessageInfo(mi)
732
+
}
733
+
734
+
func (x *StatusEvent) String() string {
735
+
return protoimpl.X.MessageStringOf(x)
736
+
}
737
+
738
+
func (*StatusEvent) ProtoMessage() {}
739
+
740
+
func (x *StatusEvent) ProtoReflect() protoreflect.Message {
741
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[11]
742
+
if x != nil {
743
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
744
+
if ms.LoadMessageInfo() == nil {
745
+
ms.StoreMessageInfo(mi)
746
+
}
747
+
return ms
748
+
}
749
+
return mi.MessageOf(x)
750
+
}
751
+
752
+
// Deprecated: Use StatusEvent.ProtoReflect.Descriptor instead.
753
+
func (*StatusEvent) Descriptor() ([]byte, []int) {
754
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{11}
755
+
}
756
+
757
+
func (x *StatusEvent) GetOffset() uint64 {
758
+
if x != nil {
759
+
return x.Offset
760
+
}
761
+
return 0
762
+
}
763
+
764
+
func (x *StatusEvent) GetLeaseId() string {
765
+
if x != nil {
766
+
return x.LeaseId
767
+
}
768
+
return ""
769
+
}
770
+
771
+
func (x *StatusEvent) GetStatus() string {
772
+
if x != nil {
773
+
return x.Status
774
+
}
775
+
return ""
776
+
}
777
+
778
+
func (x *StatusEvent) GetError() string {
779
+
if x != nil {
780
+
return x.Error
781
+
}
782
+
return ""
783
+
}
784
+
785
+
func (x *StatusEvent) GetExitCode() int64 {
786
+
if x != nil {
787
+
return x.ExitCode
788
+
}
789
+
return 0
790
+
}
791
+
792
+
func (x *StatusEvent) GetWorkflow() string {
793
+
if x != nil {
794
+
return x.Workflow
795
+
}
796
+
return ""
797
+
}
798
+
799
+
func (x *StatusEvent) GetPipelineAturi() string {
800
+
if x != nil {
801
+
return x.PipelineAturi
802
+
}
803
+
return ""
804
+
}
805
+
806
+
// LogLine relays one already-encoded models.LogLine JSON line. Carries an
807
+
// offset on the same per-session sequence as StatusEvent.
808
+
type LogLine struct {
809
+
state protoimpl.MessageState `protogen:"open.v1"`
810
+
Offset uint64 `protobuf:"varint,1,opt,name=offset,proto3" json:"offset,omitempty"`
811
+
LeaseId string `protobuf:"bytes,2,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"`
812
+
RawJson []byte `protobuf:"bytes,3,opt,name=raw_json,json=rawJson,proto3" json:"raw_json,omitempty"`
813
+
unknownFields protoimpl.UnknownFields
814
+
sizeCache protoimpl.SizeCache
815
+
}
816
+
817
+
func (x *LogLine) Reset() {
818
+
*x = LogLine{}
819
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[12]
820
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
821
+
ms.StoreMessageInfo(mi)
822
+
}
823
+
824
+
func (x *LogLine) String() string {
825
+
return protoimpl.X.MessageStringOf(x)
826
+
}
827
+
828
+
func (*LogLine) ProtoMessage() {}
829
+
830
+
func (x *LogLine) ProtoReflect() protoreflect.Message {
831
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[12]
832
+
if x != nil {
833
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
834
+
if ms.LoadMessageInfo() == nil {
835
+
ms.StoreMessageInfo(mi)
836
+
}
837
+
return ms
838
+
}
839
+
return mi.MessageOf(x)
840
+
}
841
+
842
+
// Deprecated: Use LogLine.ProtoReflect.Descriptor instead.
843
+
func (*LogLine) Descriptor() ([]byte, []int) {
844
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{12}
845
+
}
846
+
847
+
func (x *LogLine) GetOffset() uint64 {
848
+
if x != nil {
849
+
return x.Offset
850
+
}
851
+
return 0
852
+
}
853
+
854
+
func (x *LogLine) GetLeaseId() string {
855
+
if x != nil {
856
+
return x.LeaseId
857
+
}
858
+
return ""
859
+
}
860
+
861
+
func (x *LogLine) GetRawJson() []byte {
862
+
if x != nil {
863
+
return x.RawJson
864
+
}
865
+
return nil
866
+
}
867
+
868
+
// AttemptResult is the terminal signal that wakes the mill's blocked RunStep
869
+
// and resolves the lease. terminal_status is one of success/failed/timeout/
870
+
// cancelled. Carries an offset on the same per-session sequence.
871
+
type AttemptResult struct {
872
+
state protoimpl.MessageState `protogen:"open.v1"`
873
+
Offset uint64 `protobuf:"varint,1,opt,name=offset,proto3" json:"offset,omitempty"`
874
+
LeaseId string `protobuf:"bytes,2,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"`
875
+
TerminalStatus string `protobuf:"bytes,3,opt,name=terminal_status,json=terminalStatus,proto3" json:"terminal_status,omitempty"`
876
+
Error string `protobuf:"bytes,4,opt,name=error,proto3" json:"error,omitempty"`
877
+
ExitCode int64 `protobuf:"varint,5,opt,name=exit_code,json=exitCode,proto3" json:"exit_code,omitempty"`
878
+
unknownFields protoimpl.UnknownFields
879
+
sizeCache protoimpl.SizeCache
880
+
}
881
+
882
+
func (x *AttemptResult) Reset() {
883
+
*x = AttemptResult{}
884
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[13]
885
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
886
+
ms.StoreMessageInfo(mi)
887
+
}
888
+
889
+
func (x *AttemptResult) String() string {
890
+
return protoimpl.X.MessageStringOf(x)
891
+
}
892
+
893
+
func (*AttemptResult) ProtoMessage() {}
894
+
895
+
func (x *AttemptResult) ProtoReflect() protoreflect.Message {
896
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[13]
897
+
if x != nil {
898
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
899
+
if ms.LoadMessageInfo() == nil {
900
+
ms.StoreMessageInfo(mi)
901
+
}
902
+
return ms
903
+
}
904
+
return mi.MessageOf(x)
905
+
}
906
+
907
+
// Deprecated: Use AttemptResult.ProtoReflect.Descriptor instead.
908
+
func (*AttemptResult) Descriptor() ([]byte, []int) {
909
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{13}
910
+
}
911
+
912
+
func (x *AttemptResult) GetOffset() uint64 {
913
+
if x != nil {
914
+
return x.Offset
915
+
}
916
+
return 0
917
+
}
918
+
919
+
func (x *AttemptResult) GetLeaseId() string {
920
+
if x != nil {
921
+
return x.LeaseId
922
+
}
923
+
return ""
924
+
}
925
+
926
+
func (x *AttemptResult) GetTerminalStatus() string {
927
+
if x != nil {
928
+
return x.TerminalStatus
929
+
}
930
+
return ""
931
+
}
932
+
933
+
func (x *AttemptResult) GetError() string {
934
+
if x != nil {
935
+
return x.Error
936
+
}
937
+
return ""
938
+
}
939
+
940
+
func (x *AttemptResult) GetExitCode() int64 {
941
+
if x != nil {
942
+
return x.ExitCode
943
+
}
944
+
return 0
945
+
}
946
+
947
+
// Ack tells the executor the mill has durably processed all relay entries up
948
+
// to and including up_to_offset, so it may trim its buffer.
949
+
type Ack struct {
950
+
state protoimpl.MessageState `protogen:"open.v1"`
951
+
UpToOffset uint64 `protobuf:"varint,1,opt,name=up_to_offset,json=upToOffset,proto3" json:"up_to_offset,omitempty"`
952
+
unknownFields protoimpl.UnknownFields
953
+
sizeCache protoimpl.SizeCache
954
+
}
955
+
956
+
func (x *Ack) Reset() {
957
+
*x = Ack{}
958
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[14]
959
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
960
+
ms.StoreMessageInfo(mi)
961
+
}
962
+
963
+
func (x *Ack) String() string {
964
+
return protoimpl.X.MessageStringOf(x)
965
+
}
966
+
967
+
func (*Ack) ProtoMessage() {}
968
+
969
+
func (x *Ack) ProtoReflect() protoreflect.Message {
970
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[14]
971
+
if x != nil {
972
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
973
+
if ms.LoadMessageInfo() == nil {
974
+
ms.StoreMessageInfo(mi)
975
+
}
976
+
return ms
977
+
}
978
+
return mi.MessageOf(x)
979
+
}
980
+
981
+
// Deprecated: Use Ack.ProtoReflect.Descriptor instead.
982
+
func (*Ack) Descriptor() ([]byte, []int) {
983
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{14}
984
+
}
985
+
986
+
func (x *Ack) GetUpToOffset() uint64 {
987
+
if x != nil {
988
+
return x.UpToOffset
989
+
}
990
+
return 0
991
+
}
992
+
993
+
type Message struct {
994
+
state protoimpl.MessageState `protogen:"open.v1"`
995
+
Hello *Hello `protobuf:"bytes,1,opt,name=hello,proto3" json:"hello,omitempty"`
996
+
Resume *Resume `protobuf:"bytes,2,opt,name=resume,proto3" json:"resume,omitempty"`
997
+
NodeSnapshot *NodeSnapshot `protobuf:"bytes,3,opt,name=node_snapshot,json=nodeSnapshot,proto3" json:"node_snapshot,omitempty"`
998
+
ReserveSeat *ReserveSeat `protobuf:"bytes,4,opt,name=reserve_seat,json=reserveSeat,proto3" json:"reserve_seat,omitempty"`
999
+
ReserveResult *ReserveResult `protobuf:"bytes,5,opt,name=reserve_result,json=reserveResult,proto3" json:"reserve_result,omitempty"`
1000
+
CommitLease *CommitLease `protobuf:"bytes,6,opt,name=commit_lease,json=commitLease,proto3" json:"commit_lease,omitempty"`
1001
+
Committed *Committed `protobuf:"bytes,7,opt,name=committed,proto3" json:"committed,omitempty"`
1002
+
ReleaseLease *ReleaseLease `protobuf:"bytes,8,opt,name=release_lease,json=releaseLease,proto3" json:"release_lease,omitempty"`
1003
+
CancelAttempt *CancelAttempt `protobuf:"bytes,9,opt,name=cancel_attempt,json=cancelAttempt,proto3" json:"cancel_attempt,omitempty"`
1004
+
StatusEvent *StatusEvent `protobuf:"bytes,10,opt,name=status_event,json=statusEvent,proto3" json:"status_event,omitempty"`
1005
+
LogLine *LogLine `protobuf:"bytes,11,opt,name=log_line,json=logLine,proto3" json:"log_line,omitempty"`
1006
+
AttemptResult *AttemptResult `protobuf:"bytes,12,opt,name=attempt_result,json=attemptResult,proto3" json:"attempt_result,omitempty"`
1007
+
Ack *Ack `protobuf:"bytes,13,opt,name=ack,proto3" json:"ack,omitempty"`
1008
+
unknownFields protoimpl.UnknownFields
1009
+
sizeCache protoimpl.SizeCache
1010
+
}
1011
+
1012
+
func (x *Message) Reset() {
1013
+
*x = Message{}
1014
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[15]
1015
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1016
+
ms.StoreMessageInfo(mi)
1017
+
}
1018
+
1019
+
func (x *Message) String() string {
1020
+
return protoimpl.X.MessageStringOf(x)
1021
+
}
1022
+
1023
+
func (*Message) ProtoMessage() {}
1024
+
1025
+
func (x *Message) ProtoReflect() protoreflect.Message {
1026
+
mi := &file_spindle_mill_v1_mill_proto_msgTypes[15]
1027
+
if x != nil {
1028
+
ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x))
1029
+
if ms.LoadMessageInfo() == nil {
1030
+
ms.StoreMessageInfo(mi)
1031
+
}
1032
+
return ms
1033
+
}
1034
+
return mi.MessageOf(x)
1035
+
}
1036
+
1037
+
// Deprecated: Use Message.ProtoReflect.Descriptor instead.
1038
+
func (*Message) Descriptor() ([]byte, []int) {
1039
+
return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{15}
1040
+
}
1041
+
1042
+
func (x *Message) GetHello() *Hello {
1043
+
if x != nil {
1044
+
return x.Hello
1045
+
}
1046
+
return nil
1047
+
}
1048
+
1049
+
func (x *Message) GetResume() *Resume {
1050
+
if x != nil {
1051
+
return x.Resume
1052
+
}
1053
+
return nil
1054
+
}
1055
+
1056
+
func (x *Message) GetNodeSnapshot() *NodeSnapshot {
1057
+
if x != nil {
1058
+
return x.NodeSnapshot
1059
+
}
1060
+
return nil
1061
+
}
1062
+
1063
+
func (x *Message) GetReserveSeat() *ReserveSeat {
1064
+
if x != nil {
1065
+
return x.ReserveSeat
1066
+
}
1067
+
return nil
1068
+
}
1069
+
1070
+
func (x *Message) GetReserveResult() *ReserveResult {
1071
+
if x != nil {
1072
+
return x.ReserveResult
1073
+
}
1074
+
return nil
1075
+
}
1076
+
1077
+
func (x *Message) GetCommitLease() *CommitLease {
1078
+
if x != nil {
1079
+
return x.CommitLease
1080
+
}
1081
+
return nil
1082
+
}
1083
+
1084
+
func (x *Message) GetCommitted() *Committed {
1085
+
if x != nil {
1086
+
return x.Committed
1087
+
}
1088
+
return nil
1089
+
}
1090
+
1091
+
func (x *Message) GetReleaseLease() *ReleaseLease {
1092
+
if x != nil {
1093
+
return x.ReleaseLease
1094
+
}
1095
+
return nil
1096
+
}
1097
+
1098
+
func (x *Message) GetCancelAttempt() *CancelAttempt {
1099
+
if x != nil {
1100
+
return x.CancelAttempt
1101
+
}
1102
+
return nil
1103
+
}
1104
+
1105
+
func (x *Message) GetStatusEvent() *StatusEvent {
1106
+
if x != nil {
1107
+
return x.StatusEvent
1108
+
}
1109
+
return nil
1110
+
}
1111
+
1112
+
func (x *Message) GetLogLine() *LogLine {
1113
+
if x != nil {
1114
+
return x.LogLine
1115
+
}
1116
+
return nil
1117
+
}
1118
+
1119
+
func (x *Message) GetAttemptResult() *AttemptResult {
1120
+
if x != nil {
1121
+
return x.AttemptResult
1122
+
}
1123
+
return nil
1124
+
}
1125
+
1126
+
func (x *Message) GetAck() *Ack {
1127
+
if x != nil {
1128
+
return x.Ack
1129
+
}
1130
+
return nil
1131
+
}
1132
+
1133
+
var File_spindle_mill_v1_mill_proto protoreflect.FileDescriptor
1134
+
1135
+
const file_spindle_mill_v1_mill_proto_rawDesc = "" +
1136
+
"\n" +
1137
+
"\x1aspindle/mill/v1/mill.proto\x12\x0fspindle.mill.v1\x1a\x1bbuf/validate/validate.proto\"\xa3\x01\n" +
1138
+
"\x05Hello\x12)\n" +
1139
+
"\x10protocol_version\x18\x01 \x01(\rR\x0fprotocolVersion\x12 \n" +
1140
+
"\anode_id\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\x06nodeId\x12\x18\n" +
1141
+
"\aengines\x18\x03 \x03(\tR\aengines\x12\x12\n" +
1142
+
"\x04arch\x18\x04 \x01(\tR\x04arch\x12\x1f\n" +
1143
+
"\vlast_offset\x18\x05 \x01(\x04R\n" +
1144
+
"lastOffset\"'\n" +
1145
+
"\x06Resume\x12\x1d\n" +
1146
+
"\n" +
1147
+
"ack_offset\x18\x01 \x01(\x04R\tackOffset\"\x9a\x01\n" +
1148
+
"\x0eEngineSnapshot\x12\x1d\n" +
1149
+
"\n" +
1150
+
"free_seats\x18\x01 \x01(\rR\tfreeSeats\x12&\n" +
1151
+
"\x0ffree_memory_mib\x18\x02 \x01(\x03R\rfreeMemoryMib\x12\x1d\n" +
1152
+
"\n" +
1153
+
"free_vcpus\x18\x03 \x01(\x03R\tfreeVcpus\x12\"\n" +
1154
+
"\rfree_disk_mib\x18\x04 \x01(\x03R\vfreeDiskMib\"\xdc\x01\n" +
1155
+
"\fNodeSnapshot\x12\x17\n" +
1156
+
"\anode_id\x18\x01 \x01(\tR\x06nodeId\x12\x10\n" +
1157
+
"\x03seq\x18\x02 \x01(\x04R\x03seq\x12D\n" +
1158
+
"\aengines\x18\x03 \x03(\v2*.spindle.mill.v1.NodeSnapshot.EnginesEntryR\aengines\x1a[\n" +
1159
+
"\fEnginesEntry\x12\x10\n" +
1160
+
"\x03key\x18\x01 \x01(\tR\x03key\x125\n" +
1161
+
"\x05value\x18\x02 \x01(\v2\x1f.spindle.mill.v1.EngineSnapshotR\x05value:\x028\x01\"\x80\x02\n" +
1162
+
"\vReserveSeat\x12\"\n" +
1163
+
"\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12,\n" +
1164
+
"\rtarget_engine\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\ftargetEngine\x12*\n" +
1165
+
"\x11raw_pipeline_json\x18\x03 \x01(\tR\x0frawPipelineJson\x12*\n" +
1166
+
"\x11raw_workflow_json\x18\x04 \x01(\tR\x0frawWorkflowJson\x12\x12\n" +
1167
+
"\x04knot\x18\x05 \x01(\tR\x04knot\x12\x12\n" +
1168
+
"\x04rkey\x18\x06 \x01(\tR\x04rkey\x12\x1f\n" +
1169
+
"\vttl_seconds\x18\a \x01(\rR\n" +
1170
+
"ttlSeconds\"\x8a\x01\n" +
1171
+
"\rReserveResult\x12\"\n" +
1172
+
"\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12\x1a\n" +
1173
+
"\baccepted\x18\x02 \x01(\bR\baccepted\x12#\n" +
1174
+
"\rreject_reason\x18\x03 \x01(\tR\frejectReason\x12\x14\n" +
1175
+
"\x05score\x18\x04 \x01(\x01R\x05score\"0\n" +
1176
+
"\x06Secret\x12\x10\n" +
1177
+
"\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" +
1178
+
"\x05value\x18\x02 \x01(\tR\x05value\"d\n" +
1179
+
"\vCommitLease\x12\"\n" +
1180
+
"\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x121\n" +
1181
+
"\asecrets\x18\x02 \x03(\v2\x17.spindle.mill.v1.SecretR\asecrets\"/\n" +
1182
+
"\tCommitted\x12\"\n" +
1183
+
"\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\"2\n" +
1184
+
"\fReleaseLease\x12\"\n" +
1185
+
"\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\"K\n" +
1186
+
"\rCancelAttempt\x12\"\n" +
1187
+
"\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12\x16\n" +
1188
+
"\x06reason\x18\x02 \x01(\tR\x06reason\"\xd7\x01\n" +
1189
+
"\vStatusEvent\x12\x16\n" +
1190
+
"\x06offset\x18\x01 \x01(\x04R\x06offset\x12\"\n" +
1191
+
"\blease_id\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12\x16\n" +
1192
+
"\x06status\x18\x03 \x01(\tR\x06status\x12\x14\n" +
1193
+
"\x05error\x18\x04 \x01(\tR\x05error\x12\x1b\n" +
1194
+
"\texit_code\x18\x05 \x01(\x03R\bexitCode\x12\x1a\n" +
1195
+
"\bworkflow\x18\x06 \x01(\tR\bworkflow\x12%\n" +
1196
+
"\x0epipeline_aturi\x18\a \x01(\tR\rpipelineAturi\"`\n" +
1197
+
"\aLogLine\x12\x16\n" +
1198
+
"\x06offset\x18\x01 \x01(\x04R\x06offset\x12\"\n" +
1199
+
"\blease_id\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12\x19\n" +
1200
+
"\braw_json\x18\x03 \x01(\fR\arawJson\"\xa7\x01\n" +
1201
+
"\rAttemptResult\x12\x16\n" +
1202
+
"\x06offset\x18\x01 \x01(\x04R\x06offset\x12\"\n" +
1203
+
"\blease_id\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12'\n" +
1204
+
"\x0fterminal_status\x18\x03 \x01(\tR\x0eterminalStatus\x12\x14\n" +
1205
+
"\x05error\x18\x04 \x01(\tR\x05error\x12\x1b\n" +
1206
+
"\texit_code\x18\x05 \x01(\x03R\bexitCode\"'\n" +
1207
+
"\x03Ack\x12 \n" +
1208
+
"\fup_to_offset\x18\x01 \x01(\x04R\n" +
1209
+
"upToOffset\"\xcc\a\n" +
1210
+
"\aMessage\x12,\n" +
1211
+
"\x05hello\x18\x01 \x01(\v2\x16.spindle.mill.v1.HelloR\x05hello\x12/\n" +
1212
+
"\x06resume\x18\x02 \x01(\v2\x17.spindle.mill.v1.ResumeR\x06resume\x12B\n" +
1213
+
"\rnode_snapshot\x18\x03 \x01(\v2\x1d.spindle.mill.v1.NodeSnapshotR\fnodeSnapshot\x12?\n" +
1214
+
"\freserve_seat\x18\x04 \x01(\v2\x1c.spindle.mill.v1.ReserveSeatR\vreserveSeat\x12E\n" +
1215
+
"\x0ereserve_result\x18\x05 \x01(\v2\x1e.spindle.mill.v1.ReserveResultR\rreserveResult\x12?\n" +
1216
+
"\fcommit_lease\x18\x06 \x01(\v2\x1c.spindle.mill.v1.CommitLeaseR\vcommitLease\x128\n" +
1217
+
"\tcommitted\x18\a \x01(\v2\x1a.spindle.mill.v1.CommittedR\tcommitted\x12B\n" +
1218
+
"\rrelease_lease\x18\b \x01(\v2\x1d.spindle.mill.v1.ReleaseLeaseR\freleaseLease\x12E\n" +
1219
+
"\x0ecancel_attempt\x18\t \x01(\v2\x1e.spindle.mill.v1.CancelAttemptR\rcancelAttempt\x12?\n" +
1220
+
"\fstatus_event\x18\n" +
1221
+
" \x01(\v2\x1c.spindle.mill.v1.StatusEventR\vstatusEvent\x123\n" +
1222
+
"\blog_line\x18\v \x01(\v2\x18.spindle.mill.v1.LogLineR\alogLine\x12E\n" +
1223
+
"\x0eattempt_result\x18\f \x01(\v2\x1e.spindle.mill.v1.AttemptResultR\rattemptResult\x12&\n" +
1224
+
"\x03ack\x18\r \x01(\v2\x14.spindle.mill.v1.AckR\x03ack:\xaa\x01\xbaH\xa6\x01\"\xa3\x01\n" +
1225
+
"\x05hello\n" +
1226
+
"\x06resume\n" +
1227
+
"\rnode_snapshot\n" +
1228
+
"\freserve_seat\n" +
1229
+
"\x0ereserve_result\n" +
1230
+
"\fcommit_lease\n" +
1231
+
"\tcommitted\n" +
1232
+
"\rrelease_lease\n" +
1233
+
"\x0ecancel_attempt\n" +
1234
+
"\fstatus_event\n" +
1235
+
"\blog_line\n" +
1236
+
"\x0eattempt_result\n" +
1237
+
"\x03ack\x10\x01B0Z.tangled.org/core/spindle/mill/proto/gen;millv1b\x06proto3"
1238
+
1239
+
var (
1240
+
file_spindle_mill_v1_mill_proto_rawDescOnce sync.Once
1241
+
file_spindle_mill_v1_mill_proto_rawDescData []byte
1242
+
)
1243
+
1244
+
func file_spindle_mill_v1_mill_proto_rawDescGZIP() []byte {
1245
+
file_spindle_mill_v1_mill_proto_rawDescOnce.Do(func() {
1246
+
file_spindle_mill_v1_mill_proto_rawDescData = protoimpl.X.CompressGZIP(unsafe.Slice(unsafe.StringData(file_spindle_mill_v1_mill_proto_rawDesc), len(file_spindle_mill_v1_mill_proto_rawDesc)))
1247
+
})
1248
+
return file_spindle_mill_v1_mill_proto_rawDescData
1249
+
}
1250
+
1251
+
var file_spindle_mill_v1_mill_proto_msgTypes = make([]protoimpl.MessageInfo, 17)
1252
+
var file_spindle_mill_v1_mill_proto_goTypes = []any{
1253
+
(*Hello)(nil), // 0: spindle.mill.v1.Hello
1254
+
(*Resume)(nil), // 1: spindle.mill.v1.Resume
1255
+
(*EngineSnapshot)(nil), // 2: spindle.mill.v1.EngineSnapshot
1256
+
(*NodeSnapshot)(nil), // 3: spindle.mill.v1.NodeSnapshot
1257
+
(*ReserveSeat)(nil), // 4: spindle.mill.v1.ReserveSeat
1258
+
(*ReserveResult)(nil), // 5: spindle.mill.v1.ReserveResult
1259
+
(*Secret)(nil), // 6: spindle.mill.v1.Secret
1260
+
(*CommitLease)(nil), // 7: spindle.mill.v1.CommitLease
1261
+
(*Committed)(nil), // 8: spindle.mill.v1.Committed
1262
+
(*ReleaseLease)(nil), // 9: spindle.mill.v1.ReleaseLease
1263
+
(*CancelAttempt)(nil), // 10: spindle.mill.v1.CancelAttempt
1264
+
(*StatusEvent)(nil), // 11: spindle.mill.v1.StatusEvent
1265
+
(*LogLine)(nil), // 12: spindle.mill.v1.LogLine
1266
+
(*AttemptResult)(nil), // 13: spindle.mill.v1.AttemptResult
1267
+
(*Ack)(nil), // 14: spindle.mill.v1.Ack
1268
+
(*Message)(nil), // 15: spindle.mill.v1.Message
1269
+
nil, // 16: spindle.mill.v1.NodeSnapshot.EnginesEntry
1270
+
}
1271
+
var file_spindle_mill_v1_mill_proto_depIdxs = []int32{
1272
+
16, // 0: spindle.mill.v1.NodeSnapshot.engines:type_name -> spindle.mill.v1.NodeSnapshot.EnginesEntry
1273
+
6, // 1: spindle.mill.v1.CommitLease.secrets:type_name -> spindle.mill.v1.Secret
1274
+
0, // 2: spindle.mill.v1.Message.hello:type_name -> spindle.mill.v1.Hello
1275
+
1, // 3: spindle.mill.v1.Message.resume:type_name -> spindle.mill.v1.Resume
1276
+
3, // 4: spindle.mill.v1.Message.node_snapshot:type_name -> spindle.mill.v1.NodeSnapshot
1277
+
4, // 5: spindle.mill.v1.Message.reserve_seat:type_name -> spindle.mill.v1.ReserveSeat
1278
+
5, // 6: spindle.mill.v1.Message.reserve_result:type_name -> spindle.mill.v1.ReserveResult
1279
+
7, // 7: spindle.mill.v1.Message.commit_lease:type_name -> spindle.mill.v1.CommitLease
1280
+
8, // 8: spindle.mill.v1.Message.committed:type_name -> spindle.mill.v1.Committed
1281
+
9, // 9: spindle.mill.v1.Message.release_lease:type_name -> spindle.mill.v1.ReleaseLease
1282
+
10, // 10: spindle.mill.v1.Message.cancel_attempt:type_name -> spindle.mill.v1.CancelAttempt
1283
+
11, // 11: spindle.mill.v1.Message.status_event:type_name -> spindle.mill.v1.StatusEvent
1284
+
12, // 12: spindle.mill.v1.Message.log_line:type_name -> spindle.mill.v1.LogLine
1285
+
13, // 13: spindle.mill.v1.Message.attempt_result:type_name -> spindle.mill.v1.AttemptResult
1286
+
14, // 14: spindle.mill.v1.Message.ack:type_name -> spindle.mill.v1.Ack
1287
+
2, // 15: spindle.mill.v1.NodeSnapshot.EnginesEntry.value:type_name -> spindle.mill.v1.EngineSnapshot
1288
+
16, // [16:16] is the sub-list for method output_type
1289
+
16, // [16:16] is the sub-list for method input_type
1290
+
16, // [16:16] is the sub-list for extension type_name
1291
+
16, // [16:16] is the sub-list for extension extendee
1292
+
0, // [0:16] is the sub-list for field type_name
1293
+
}
1294
+
1295
+
func init() { file_spindle_mill_v1_mill_proto_init() }
1296
+
func file_spindle_mill_v1_mill_proto_init() {
1297
+
if File_spindle_mill_v1_mill_proto != nil {
1298
+
return
1299
+
}
1300
+
type x struct{}
1301
+
out := protoimpl.TypeBuilder{
1302
+
File: protoimpl.DescBuilder{
1303
+
GoPackagePath: reflect.TypeOf(x{}).PkgPath(),
1304
+
RawDescriptor: unsafe.Slice(unsafe.StringData(file_spindle_mill_v1_mill_proto_rawDesc), len(file_spindle_mill_v1_mill_proto_rawDesc)),
1305
+
NumEnums: 0,
1306
+
NumMessages: 17,
1307
+
NumExtensions: 0,
1308
+
NumServices: 0,
1309
+
},
1310
+
GoTypes: file_spindle_mill_v1_mill_proto_goTypes,
1311
+
DependencyIndexes: file_spindle_mill_v1_mill_proto_depIdxs,
1312
+
MessageInfos: file_spindle_mill_v1_mill_proto_msgTypes,
1313
+
}.Build()
1314
+
File_spindle_mill_v1_mill_proto = out.File
1315
+
file_spindle_mill_v1_mill_proto_goTypes = nil
1316
+
file_spindle_mill_v1_mill_proto_depIdxs = nil
1317
+
}
+103
spindle/mill/proto/protocol.go
+103
spindle/mill/proto/protocol.go
···
1
+
// Package millproto carries the mill<->executor session protocol: a new
2
+
// message vocabulary over the same length-prefixed protobuf framing the spindle
3
+
// already uses to talk to the microVM guest (see spindle/agentproto). Only the
4
+
// framing pattern is shared; the messages are entirely separate.
5
+
package millproto
6
+
7
+
import (
8
+
"encoding/binary"
9
+
"fmt"
10
+
"io"
11
+
"sync"
12
+
13
+
"buf.build/go/protovalidate"
14
+
"google.golang.org/protobuf/proto"
15
+
16
+
millv1 "tangled.org/core/spindle/mill/proto/gen"
17
+
)
18
+
19
+
const (
20
+
ProtocolVersion = 1
21
+
// generous vs agentproto's 1 MiB: a ReserveSeat carries the raw pipeline +
22
+
// workflow JSON, and relayed log lines can be chunky.
23
+
MaxMessageBytes = 8 * 1024 * 1024
24
+
)
25
+
26
+
type Message = millv1.Message
27
+
28
+
var validator protovalidate.Validator
29
+
30
+
func init() {
31
+
var err error
32
+
validator, err = protovalidate.New()
33
+
if err != nil {
34
+
panic(fmt.Errorf("failed to initialize protovalidate validator: %w", err))
35
+
}
36
+
}
37
+
38
+
type Encoder struct {
39
+
mu sync.Mutex
40
+
w io.Writer
41
+
}
42
+
43
+
func NewEncoder(w io.Writer) *Encoder {
44
+
return &Encoder{w: w}
45
+
}
46
+
47
+
func (e *Encoder) Encode(msg *Message) error {
48
+
if err := validator.Validate(msg); err != nil {
49
+
return fmt.Errorf("validate fleet message: %w", err)
50
+
}
51
+
52
+
data, err := proto.Marshal(msg)
53
+
if err != nil {
54
+
return fmt.Errorf("marshal fleet message: %w", err)
55
+
}
56
+
if len(data) > MaxMessageBytes {
57
+
return fmt.Errorf("fleet message exceeded %d bytes", MaxMessageBytes)
58
+
}
59
+
60
+
// single write of header+payload so this maps to exactly one websocket
61
+
// binary frame when the writer is a ws stream.
62
+
frame := make([]byte, 4+len(data))
63
+
binary.BigEndian.PutUint32(frame[:4], uint32(len(data)))
64
+
copy(frame[4:], data)
65
+
66
+
e.mu.Lock()
67
+
defer e.mu.Unlock()
68
+
_, err = e.w.Write(frame)
69
+
return err
70
+
}
71
+
72
+
type Decoder struct {
73
+
r io.Reader
74
+
}
75
+
76
+
func NewDecoder(r io.Reader) *Decoder {
77
+
return &Decoder{r: r}
78
+
}
79
+
80
+
func (d *Decoder) Decode() (*Message, error) {
81
+
msg := &Message{}
82
+
var header [4]byte
83
+
if _, err := io.ReadFull(d.r, header[:]); err != nil {
84
+
return msg, err
85
+
}
86
+
87
+
size := binary.BigEndian.Uint32(header[:])
88
+
if size > MaxMessageBytes {
89
+
return msg, fmt.Errorf("fleet message exceeded %d bytes", MaxMessageBytes)
90
+
}
91
+
92
+
data := make([]byte, size)
93
+
if _, err := io.ReadFull(d.r, data); err != nil {
94
+
return msg, err
95
+
}
96
+
if err := proto.Unmarshal(data, msg); err != nil {
97
+
return msg, fmt.Errorf("parse fleet message: %w", err)
98
+
}
99
+
if err := validator.Validate(msg); err != nil {
100
+
return msg, fmt.Errorf("validate fleet message: %w", err)
101
+
}
102
+
return msg, nil
103
+
}
+69
spindle/mill/proto/protocol_test.go
+69
spindle/mill/proto/protocol_test.go
···
1
+
package millproto
2
+
3
+
import (
4
+
"bytes"
5
+
"encoding/binary"
6
+
"testing"
7
+
8
+
millv1 "tangled.org/core/spindle/mill/proto/gen"
9
+
)
10
+
11
+
func TestEncodeDecodeRoundTrip(t *testing.T) {
12
+
var buf bytes.Buffer
13
+
enc := NewEncoder(&buf)
14
+
15
+
want := &Message{
16
+
ReserveSeat: &millv1.ReserveSeat{
17
+
LeaseId: "lease-1",
18
+
TargetEngine: "microvm",
19
+
RawWorkflowJson: `{"name":"build"}`,
20
+
Knot: "knot.example",
21
+
Rkey: "abc123",
22
+
TtlSeconds: 30,
23
+
},
24
+
}
25
+
if err := enc.Encode(want); err != nil {
26
+
t.Fatalf("Encode() error = %v", err)
27
+
}
28
+
29
+
got, err := NewDecoder(&buf).Decode()
30
+
if err != nil {
31
+
t.Fatalf("Decode() error = %v", err)
32
+
}
33
+
rs := got.GetReserveSeat()
34
+
if rs == nil {
35
+
t.Fatal("decoded message missing reserve_seat")
36
+
}
37
+
if rs.LeaseId != "lease-1" || rs.TargetEngine != "microvm" || rs.TtlSeconds != 30 {
38
+
t.Fatalf("round-trip mismatch: %+v", rs)
39
+
}
40
+
}
41
+
42
+
func TestDecoderRejectsOversizedMessage(t *testing.T) {
43
+
var tooLarge bytes.Buffer
44
+
var header [4]byte
45
+
binary.BigEndian.PutUint32(header[:], MaxMessageBytes+1)
46
+
tooLarge.Write(header[:])
47
+
48
+
if _, err := NewDecoder(&tooLarge).Decode(); err == nil {
49
+
t.Fatal("expected oversized message error")
50
+
}
51
+
}
52
+
53
+
func TestValidationOneofRequired(t *testing.T) {
54
+
// exactly one payload set: ok
55
+
if err := validator.Validate(&Message{Ack: &millv1.Ack{UpToOffset: 5}}); err != nil {
56
+
t.Fatalf("expected valid message to pass, got: %v", err)
57
+
}
58
+
// zero payloads: fail
59
+
if err := validator.Validate(&Message{}); err == nil {
60
+
t.Fatal("expected message with zero payloads to fail validation")
61
+
}
62
+
// two payloads: fail
63
+
if err := validator.Validate(&Message{
64
+
Ack: &millv1.Ack{},
65
+
Committed: &millv1.Committed{LeaseId: "x"},
66
+
}); err == nil {
67
+
t.Fatal("expected message with multiple payloads to fail validation")
68
+
}
69
+
}
+169
spindle/mill/proto/spindle/mill/v1/mill.proto
+169
spindle/mill/proto/spindle/mill/v1/mill.proto
···
1
+
syntax = "proto3";
2
+
3
+
package spindle.mill.v1;
4
+
5
+
import "buf/validate/validate.proto";
6
+
7
+
option go_package = "tangled.org/core/spindle/mill/proto/gen;millv1";
8
+
9
+
// Hello is the first frame an executor sends after dialing the mill. It
10
+
// carries the static traits of the node plus a resume hint.
11
+
message Hello {
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
+
// GOARCH of the node, so the mill won't place arch-incompatible jobs.
19
+
string arch = 4;
20
+
// the highest relay offset the executor believes it has sent; a resume hint.
21
+
uint64 last_offset = 5;
22
+
}
23
+
24
+
// Resume is the mill's reply to Hello. The executor replays every buffered
25
+
// relay entry with offset strictly greater than ack_offset before sending new
26
+
// ones.
27
+
message Resume {
28
+
uint64 ack_offset = 1;
29
+
}
30
+
31
+
// EngineSnapshot is the changing per-engine state on a node. free_seats is a
32
+
// coarse "can you take more" hint; 0 also means draining (the mill treats a
33
+
// draining node as a full one until it leaves). The resource fields are coarse
34
+
// budget headroom for ranking only; the executor makes the real yes/no call in
35
+
// ReserveResult.
36
+
message EngineSnapshot {
37
+
uint32 free_seats = 1;
38
+
int64 free_memory_mib = 2;
39
+
int64 free_vcpus = 3;
40
+
int64 free_disk_mib = 4;
41
+
}
42
+
43
+
// NodeSnapshot is pushed on connect, periodically, and right after any state
44
+
// change (reserve, commit, terminal).
45
+
message NodeSnapshot {
46
+
string node_id = 1;
47
+
uint64 seq = 2;
48
+
map<string, EngineSnapshot> engines = 3;
49
+
}
50
+
51
+
// ReserveSeat asks an executor to hold a seat for a job. Zero secrets ride this
52
+
// message; the raw pipeline/workflow are carried as JSON since processPipeline
53
+
// already round-trips them through JSON.
54
+
message ReserveSeat {
55
+
string lease_id = 1 [(buf.validate.field).string.min_len = 1];
56
+
string target_engine = 2 [(buf.validate.field).string.min_len = 1];
57
+
string raw_pipeline_json = 3;
58
+
string raw_workflow_json = 4;
59
+
// the pipeline id (knot + rkey); the executor reconstructs the exact
60
+
// WorkflowId so its relayed status rows and log path match what the mill
61
+
// authored for "pending".
62
+
string knot = 5;
63
+
string rkey = 6;
64
+
uint32 ttl_seconds = 7;
65
+
}
66
+
67
+
// ReserveResult is the executor's accept/reject for a ReserveSeat.
68
+
message ReserveResult {
69
+
string lease_id = 1 [(buf.validate.field).string.min_len = 1];
70
+
bool accepted = 2;
71
+
string reject_reason = 3;
72
+
// optional bid score the mill ranks accepted leases by (higher is better).
73
+
double score = 4;
74
+
}
75
+
76
+
// Secret is a single unlocked secret, sent only inside CommitLease.
77
+
message Secret {
78
+
string key = 1;
79
+
string value = 2;
80
+
}
81
+
82
+
// CommitLease promotes a reservation to a running job and hands over the
83
+
// secrets. The executor then runs the real engine with the slot it already
84
+
// holds.
85
+
message CommitLease {
86
+
string lease_id = 1 [(buf.validate.field).string.min_len = 1];
87
+
repeated Secret secrets = 2;
88
+
}
89
+
90
+
// Committed acks a CommitLease.
91
+
message Committed {
92
+
string lease_id = 1 [(buf.validate.field).string.min_len = 1];
93
+
}
94
+
95
+
// ReleaseLease tells the executor to drop a reservation it never committed (the
96
+
// mill picked another node, or is cleaning up).
97
+
message ReleaseLease {
98
+
string lease_id = 1 [(buf.validate.field).string.min_len = 1];
99
+
}
100
+
101
+
// CancelAttempt cancels a running attempt (user cancel / DestroyWorkflow).
102
+
message CancelAttempt {
103
+
string lease_id = 1 [(buf.validate.field).string.min_len = 1];
104
+
string reason = 2;
105
+
}
106
+
107
+
// StatusEvent relays a non-terminal status (i.e. running) the executor wrote to
108
+
// its own eventstream. Terminals are authored by the mill from the attempt
109
+
// result, so they are never relayed here. Carries a per-session monotonic
110
+
// offset for gap-free replay across reconnects.
111
+
message StatusEvent {
112
+
uint64 offset = 1;
113
+
string lease_id = 2 [(buf.validate.field).string.min_len = 1];
114
+
string status = 3;
115
+
string error = 4;
116
+
int64 exit_code = 5;
117
+
string workflow = 6;
118
+
string pipeline_aturi = 7;
119
+
}
120
+
121
+
// LogLine relays one already-encoded models.LogLine JSON line. Carries an
122
+
// offset on the same per-session sequence as StatusEvent.
123
+
message LogLine {
124
+
uint64 offset = 1;
125
+
string lease_id = 2 [(buf.validate.field).string.min_len = 1];
126
+
bytes raw_json = 3;
127
+
}
128
+
129
+
// AttemptResult is the terminal signal that wakes the mill's blocked RunStep
130
+
// and resolves the lease. terminal_status is one of success/failed/timeout/
131
+
// cancelled. Carries an offset on the same per-session sequence.
132
+
message AttemptResult {
133
+
uint64 offset = 1;
134
+
string lease_id = 2 [(buf.validate.field).string.min_len = 1];
135
+
string terminal_status = 3;
136
+
string error = 4;
137
+
int64 exit_code = 5;
138
+
}
139
+
140
+
// Ack tells the executor the mill has durably processed all relay entries up
141
+
// to and including up_to_offset, so it may trim its buffer.
142
+
message Ack {
143
+
uint64 up_to_offset = 1;
144
+
}
145
+
146
+
message Message {
147
+
option (buf.validate.message).oneof = {
148
+
fields: [
149
+
"hello", "resume", "node_snapshot", "reserve_seat", "reserve_result",
150
+
"commit_lease", "committed", "release_lease", "cancel_attempt",
151
+
"status_event", "log_line", "attempt_result", "ack"
152
+
],
153
+
required: true
154
+
};
155
+
156
+
Hello hello = 1;
157
+
Resume resume = 2;
158
+
NodeSnapshot node_snapshot = 3;
159
+
ReserveSeat reserve_seat = 4;
160
+
ReserveResult reserve_result = 5;
161
+
CommitLease commit_lease = 6;
162
+
Committed committed = 7;
163
+
ReleaseLease release_lease = 8;
164
+
CancelAttempt cancel_attempt = 9;
165
+
StatusEvent status_event = 10;
166
+
LogLine log_line = 11;
167
+
AttemptResult attempt_result = 12;
168
+
Ack ack = 13;
169
+
}
+58
spindle/mill/proto/ws.go
+58
spindle/mill/proto/ws.go
···
1
+
package millproto
2
+
3
+
import (
4
+
"io"
5
+
"sync"
6
+
7
+
"github.com/gorilla/websocket"
8
+
)
9
+
10
+
// WSStream adapts a gorilla websocket connection to an io.ReadWriteCloser so the
11
+
// length-prefixed fleet framing rides over it. Each Encode produces exactly one
12
+
// binary frame; the reader reassembles the byte stream across frames.
13
+
type WSStream struct {
14
+
conn *websocket.Conn
15
+
16
+
rmu sync.Mutex
17
+
r io.Reader // current message reader, advanced as frames are consumed
18
+
19
+
wmu sync.Mutex
20
+
}
21
+
22
+
func NewWSStream(conn *websocket.Conn) *WSStream {
23
+
return &WSStream{conn: conn}
24
+
}
25
+
26
+
func (s *WSStream) Read(p []byte) (int, error) {
27
+
s.rmu.Lock()
28
+
defer s.rmu.Unlock()
29
+
for {
30
+
if s.r == nil {
31
+
_, r, err := s.conn.NextReader()
32
+
if err != nil {
33
+
return 0, err
34
+
}
35
+
s.r = r
36
+
}
37
+
n, err := s.r.Read(p)
38
+
if err == io.EOF {
39
+
s.r = nil
40
+
if n > 0 {
41
+
return n, nil
42
+
}
43
+
continue
44
+
}
45
+
return n, err
46
+
}
47
+
}
48
+
49
+
func (s *WSStream) Write(p []byte) (int, error) {
50
+
s.wmu.Lock()
51
+
defer s.wmu.Unlock()
52
+
if err := s.conn.WriteMessage(websocket.BinaryMessage, p); err != nil {
53
+
return 0, err
54
+
}
55
+
return len(p), nil
56
+
}
57
+
58
+
func (s *WSStream) Close() error { return s.conn.Close() }