+74 -61

spindle/mill/executor/executor.go

Reviewed

··· 3 3 import ( 4 4 "context" 5 5 "encoding/json" 6 6 + "errors" 6 7 "fmt" 7 8 "log/slog" 8 9 "maps" 9 10 "net/http" 10 11 "runtime" 12 12 + "strings" 11 13 "sync" 12 14 "time" 13 15 ··· 34 36 35 37 type Executor struct { 36 38 millURL string 37 37 - secret string 39 39 + token string 38 40 nodeID string 39 41 seats int 42 42 + labels []string 40 43 41 44 engines map[string]models.Engine 42 45 db *db.DB ··· 50 53 connMu sync.Mutex 51 54 enc messageEncoder 52 55 53 53 - mu sync.Mutex 54 54 - active map[string]*reservation 55 55 - draining bool 56 56 - activeCount int 56 56 + mu sync.Mutex 57 57 + active map[string]*reservation 58 58 + draining bool 57 59 } 58 60 59 61 type reservation struct { ··· 75 77 Encode(*millproto.Message) error 76 78 } 77 79 78 78 - type reservationCleanup struct { 79 79 - stopTail func() 80 80 - slot engine.WorkflowSlot 81 81 - } 82 82 - 83 83 - func (c reservationCleanup) run() { 84 84 - if c.stopTail != nil { 85 85 - c.stopTail() 86 86 - } 87 87 - if c.slot != nil { 88 88 - c.slot.Release() 89 89 - } 90 90 - } 91 91 - 92 80 func New(cfg *config.Config, engines map[string]models.Engine, d *db.DB, n *notifier.Notifier, l *slog.Logger) *Executor { 93 81 seats := cfg.Mill.Seats 94 82 if seats <= 0 { 95 83 seats = defaultSeats 96 84 } 85 85 + labels := normalizeLabels(cfg.Mill.Labels) 97 86 return &Executor{ 98 87 millURL: cfg.Mill.URL, 99 99 - secret: cfg.Mill.SharedSecret, 88 88 + token: cfg.Mill.SharedSecret, 100 89 nodeID: cfg.Server.Hostname, 101 90 seats: seats, 91 91 + labels: labels, 102 92 engines: engines, 103 93 db: d, 104 94 n: n, ··· 142 132 143 133 func (e *Executor) runSession(ctx context.Context) error { 144 134 header := http.Header{} 145 145 - if e.secret != "" { 146 146 - header.Set("Authorization", "Bearer "+e.secret) 135 135 + if e.token != "" { 136 136 + header.Set("Authorization", "Bearer "+e.token) 147 137 } 148 138 conn, _, err := websocket.DefaultDialer.DialContext(ctx, e.millURL, header) 149 139 if err != nil { ··· 165 155 NodeId: e.nodeID, 166 156 Engines: e.engineNames(), 167 157 Arch: runtime.GOARCH, 158 158 + Labels: e.labels, 168 159 LastOffset: e.relay.lastOffset(), 169 160 }} 170 161 if err := enc.Encode(hello); err != nil { ··· 248 239 // --- reserve / commit / release / cancel ---------------------------------- 249 240 250 241 func (e *Executor) handleReserve(ctx context.Context, rs *millv1.ReserveSeat) { 251 251 - reject := func(reason string) { 242 242 + reject := func(reason string, class millv1.RejectClass) { 252 243 e.send(&millproto.Message{ReserveResult: &millv1.ReserveResult{ 253 244 LeaseId: rs.GetLeaseId(), 254 245 Accepted: false, 255 246 RejectReason: reason, 247 247 + RejectClass: class, 256 248 }}) 257 249 } 258 250 ··· 260 252 draining := e.draining 261 253 e.mu.Unlock() 262 254 if draining { 263 263 - reject("draining") 255 255 + reject("draining", millv1.RejectClass_REJECT_CLASS_TRANSIENT) 264 256 return 265 257 } 266 258 267 259 realEngine, ok := e.engines[rs.GetTargetEngine()] 268 260 if !ok { 269 269 - reject("unknown engine " + rs.GetTargetEngine()) 261 261 + reject("unknown engine "+rs.GetTargetEngine(), millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE) 270 262 return 271 263 } 272 264 slotter, ok := realEngine.(engine.WorkflowSlotter) 273 265 if !ok { 274 274 - reject("engine does not support workflow slots") 266 266 + reject("engine does not support workflow slots", millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE) 275 267 return 276 268 } 277 269 278 270 var twf tangled.Pipeline_Workflow 279 271 if err := json.Unmarshal([]byte(rs.GetRawWorkflowJson()), &twf); err != nil { 280 280 - reject("bad workflow json") 272 272 + reject("bad workflow json", millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE) 281 273 return 282 274 } 283 275 var tpl tangled.Pipeline 284 276 if err := json.Unmarshal([]byte(rs.GetRawPipelineJson()), &tpl); err != nil { 285 285 - reject("bad pipeline json") 277 277 + reject("bad pipeline json", millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE) 286 278 return 287 279 } 288 280 ··· 291 283 292 284 wf, err := realEngine.InitWorkflow(twf, tpl) 293 285 if err != nil { 294 294 - reject("init workflow: " + err.Error()) 286 286 + reject("init workflow: "+err.Error(), millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE) 295 287 return 296 288 } 297 289 // the job skipped processPipeline, so inject TANGLED_* env here. ··· 303 295 // NoWait: the executor doesn't queue locally; the mill owns the backlog. 304 296 slot, err := slotter.AcquireWorkflowSlot(ctx, wid, wf, engine.NoWait) 305 297 if err != nil { 306 306 - reject(err.Error()) 298 298 + class := millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE 299 299 + if errors.Is(err, engine.ErrNoWorkflowSlots) { 300 300 + class = millv1.RejectClass_REJECT_CLASS_TRANSIENT 301 301 + } 302 302 + reject(err.Error(), class) 307 303 return 308 304 } 309 305 ··· 324 320 325 321 e.mu.Lock() 326 322 e.active[res.leaseID] = res 327 327 - e.activeCount++ 328 323 e.mu.Unlock() 329 324 330 325 e.send(&millproto.Message{ReserveResult: &millv1.ReserveResult{ 331 326 LeaseId: rs.GetLeaseId(), 332 327 Accepted: true, 333 333 - Score: e.bidScore(), 334 328 }}) 335 329 e.pushSnapshot() 336 330 } ··· 380 374 if !ok { 381 375 return 382 376 } 383 383 - cleanup.run() 377 377 + cleanup() 384 378 e.pushSnapshot() 385 379 } 386 380 ··· 394 388 res.cancelled = true 395 389 cancel := res.cancel 396 390 committed := res.committed 397 397 - var cleanup reservationCleanup 391 391 + var cleanup func() 398 392 if !committed { 399 393 cleanup = e.removeReservationLocked(res, true) 400 394 } ··· 406 400 // committed jobs clean up when the observe loop sees the terminal row; an 407 401 // uncommitted cancel still needs the slot released. 408 402 if !committed { 409 409 - cleanup.run() 403 403 + cleanup() 410 404 e.pushSnapshot() 411 405 } 412 406 } ··· 417 411 return 418 412 } 419 413 e.l.Warn("reservation expired before commit", "lease", leaseID) 420 420 - cleanup.run() 414 414 + cleanup() 421 415 e.pushSnapshot() 422 416 } 423 417 424 424 - func (e *Executor) takeUncommittedReservation(leaseID string, releaseSlot bool) (reservationCleanup, bool) { 418 418 + func (e *Executor) takeUncommittedReservation(leaseID string, releaseSlot bool) (func(), bool) { 425 419 e.mu.Lock() 426 420 defer e.mu.Unlock() 427 421 res := e.active[leaseID] 428 422 if res == nil || res.committed { 429 429 - return reservationCleanup{}, false 423 423 + return func() {}, false 430 424 } 431 425 return e.removeReservationLocked(res, releaseSlot), true 432 426 } 433 427 434 434 - func (e *Executor) finishReservation(res *reservation) (reservationCleanup, bool, bool) { 428 428 + func (e *Executor) finishReservation(res *reservation) (func(), bool, bool) { 435 429 e.mu.Lock() 436 430 defer e.mu.Unlock() 437 431 if e.active[res.leaseID] != res { 438 438 - return reservationCleanup{}, false, false 432 432 + return func() {}, false, false 439 433 } 440 434 cancelled := res.cancelled 441 435 return e.removeReservationLocked(res, false), cancelled, true 442 436 } 443 437 444 444 - func (e *Executor) removeReservationLocked(res *reservation, releaseSlot bool) reservationCleanup { 438 438 + func (e *Executor) removeReservationLocked(res *reservation, releaseSlot bool) func() { 445 439 delete(e.active, res.leaseID) 446 446 - if e.activeCount > 0 { 447 447 - e.activeCount-- 448 448 - } 449 440 if res.ttlTimer != nil { 450 441 res.ttlTimer.Stop() 451 442 res.ttlTimer = nil 452 443 } 453 453 - cleanup := reservationCleanup{stopTail: res.stopTail} 444 444 + stopTail := res.stopTail 454 445 res.stopTail = nil 446 446 + slot := res.slot 455 447 if releaseSlot { 456 456 - cleanup.slot = res.slot 457 448 res.slot = nil 458 449 } 459 459 - return cleanup 450 450 + return func() { 451 451 + if stopTail != nil { 452 452 + stopTail() 453 453 + } 454 454 + if releaseSlot && slot != nil { 455 455 + slot.Release() 456 456 + } 457 457 + } 458 458 + } 459 459 + 460 460 + func normalizeLabels(labels []string) []string { 461 461 + seen := make(map[string]struct{}, len(labels)) 462 462 + out := make([]string, 0, len(labels)) 463 463 + for _, label := range labels { 464 464 + label = strings.TrimSpace(label) 465 465 + if label == "" { 466 466 + continue 467 467 + } 468 468 + if _, ok := seen[label]; ok { 469 469 + continue 470 470 + } 471 471 + seen[label] = struct{}{} 472 472 + out = append(out, label) 473 473 + } 474 474 + return out 460 475 } 461 476 462 477 // --- snapshots ------------------------------------------------------------- ··· 483 498 484 499 func (e *Executor) pushSnapshot() { 485 500 e.mu.Lock() 486 486 - free := uint32(0) 487 487 - if !e.draining { 488 488 - if n := e.seats - e.activeCount; n > 0 { 489 489 - free = uint32(n) 490 490 - } 491 491 - } 501 501 + draining := e.draining 502 502 + active := len(e.active) 492 503 e.mu.Unlock() 493 504 494 494 - engines := make(map[string]*millv1.EngineSnapshot, len(e.engines)) 505 505 + load := 0.0 506 506 + if e.seats > 0 { 507 507 + load = float64(active) / float64(e.seats) 508 508 + } 509 509 + available := !draining && active < e.seats 510 510 + engines := make(map[string]*millv1.EngineAvailability, len(e.engines)) 495 511 for name := range e.engines { 496 496 - engines[name] = &millv1.EngineSnapshot{FreeSeats: free} 512 512 + engines[name] = &millv1.EngineAvailability{ 513 513 + Available: available, 514 514 + Load: map[string]float64{"slots": load}, 515 515 + } 497 516 } 498 517 e.send(&millproto.Message{NodeSnapshot: &millv1.NodeSnapshot{ 499 518 NodeId: e.nodeID, ··· 506 525 e.draining = true 507 526 e.mu.Unlock() 508 527 e.pushSnapshot() 509 509 - } 510 510 - 511 511 - func (e *Executor) bidScore() float64 { 512 512 - e.mu.Lock() 513 513 - defer e.mu.Unlock() 514 514 - return float64(e.seats - e.activeCount) 515 528 } 516 529 517 530 func (e *Executor) engineNames() []string {

+1 -1

spindle/mill/executor/observe.go

Reviewed

··· 83 83 if !ok { 84 84 return 85 85 } 86 86 - cleanup.run() 86 86 + cleanup() 87 87 88 88 e.relayMu.Lock() 89 89 terminalStatus := st.Status

+2 -25

spindle/mill/executor/reserved.go

Reviewed

··· 4 4 "context" 5 5 "fmt" 6 6 "sync" 7 7 - "time" 8 7 9 9 - "tangled.org/core/api/tangled" 10 8 "tangled.org/core/spindle/engine" 11 9 "tangled.org/core/spindle/models" 12 12 - "tangled.org/core/spindle/secrets" 13 10 ) 14 11 15 12 // reservedEngine wraps a real engine so that the slot acquired up-front during ··· 19 16 // 20 17 // This is the only change to the execution path on an executor. 21 18 type reservedEngine struct { 22 22 - inner models.Engine 19 19 + models.Engine 23 20 slot engine.WorkflowSlot 24 21 once sync.Once 25 22 } ··· 27 24 // newReservedEngine returns a wrapper around inner that hands back slot exactly 28 25 // once from AcquireWorkflowSlot. 29 26 func newReservedEngine(inner models.Engine, slot engine.WorkflowSlot) models.Engine { 30 30 - return &reservedEngine{inner: inner, slot: slot} 31 31 - } 32 32 - 33 33 - func (e *reservedEngine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) { 34 34 - return e.inner.InitWorkflow(twf, tpl) 35 35 - } 36 36 - 37 37 - func (e *reservedEngine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error { 38 38 - return e.inner.SetupWorkflow(ctx, wid, wf, wfLogger) 39 39 - } 40 40 - 41 41 - func (e *reservedEngine) WorkflowTimeout() time.Duration { 42 42 - return e.inner.WorkflowTimeout() 43 43 - } 44 44 - 45 45 - func (e *reservedEngine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error { 46 46 - return e.inner.DestroyWorkflow(ctx, wid) 47 47 - } 48 48 - 49 49 - func (e *reservedEngine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, secrets []secrets.UnlockedSecret, wfLogger models.WorkflowLogger) error { 50 50 - return e.inner.RunStep(ctx, wid, w, idx, secrets, wfLogger) 27 27 + return &reservedEngine{Engine: inner, slot: slot} 51 28 } 52 29 53 30 // AcquireWorkflowSlot hands back the pre-acquired slot exactly once. The slot

+127 -47

spindle/mill/mill.go

Reviewed

··· 6 6 "errors" 7 7 "fmt" 8 8 "log/slog" 9 9 - "sort" 9 9 + "slices" 10 10 + "strings" 10 11 "sync" 11 12 "time" 12 13 ··· 274 275 return nil, err 275 276 } 276 277 277 277 - candidates := m.rankCandidates(engineName) 278 278 + requiredLabels := requiredLabels(wf) 279 279 + candidates := m.rankCandidates(engineName, requiredLabels) 278 280 if len(candidates) == 0 { 279 281 return nil, nil 280 280 - } 281 281 - if len(candidates) > m.cfg.TopK { 282 282 - candidates = candidates[:m.cfg.TopK] 283 282 } 284 283 285 285 - bidCtx, cancel := context.WithTimeout(ctx, m.cfg.BidTimeout) 286 286 - defer cancel() 287 287 - 288 284 type bidResult struct { 289 289 - sess *millSession 290 290 - lease *RemoteLease 291 291 - score float64 285 285 + sess *millSession 286 286 + lease *RemoteLease 287 287 + rank int 288 288 + incompatible bool 289 289 + reason string 292 290 } 293 291 results := make(chan bidResult, len(candidates)) 294 294 - var wg sync.WaitGroup 295 295 - for _, sess := range candidates { 296 296 - wg.Add(1) 297 297 - go func(sess *millSession) { 298 298 - defer wg.Done() 299 299 - leaseID := m.nextLeaseID() 300 300 - lease := newLease(leaseID, sess.nodeID, engineName) 301 301 - wid := m.jobWid(wf) 302 302 - msg := &millproto.Message{ReserveSeat: &millv1.ReserveSeat{ 303 303 - LeaseId: leaseID, 304 304 - TargetEngine: engineName, 305 305 - RawPipelineJson: rawPipeline, 306 306 - RawWorkflowJson: rawWorkflow, 307 307 - Knot: wid.Knot, 308 308 - Rkey: wid.Rkey, 309 309 - TtlSeconds: uint32(m.cfg.ReconnectGrace / time.Second), 310 310 - }} 311 311 - resp, err := sess.request(bidCtx, leaseID, msg) 312 312 - if err != nil { 313 313 - return 314 314 - } 315 315 - rr := resp.GetReserveResult() 316 316 - if rr == nil || !rr.GetAccepted() { 292 292 + ask := func(rank int, sess *millSession) { 293 293 + bidCtx, cancel := context.WithTimeout(ctx, m.cfg.BidTimeout) 294 294 + defer cancel() 295 295 + leaseID := m.nextLeaseID() 296 296 + lease := newLease(leaseID, sess.nodeID, engineName) 297 297 + wid := m.jobWid(wf) 298 298 + msg := &millproto.Message{ReserveSeat: &millv1.ReserveSeat{ 299 299 + LeaseId: leaseID, 300 300 + TargetEngine: engineName, 301 301 + RawPipelineJson: rawPipeline, 302 302 + RawWorkflowJson: rawWorkflow, 303 303 + Knot: wid.Knot, 304 304 + Rkey: wid.Rkey, 305 305 + TtlSeconds: uint32(m.cfg.ReconnectGrace / time.Second), 306 306 + }} 307 307 + resp, err := sess.request(bidCtx, leaseID, msg) 308 308 + if err != nil { 309 309 + results <- bidResult{} 310 310 + return 311 311 + } 312 312 + rr := resp.GetReserveResult() 313 313 + if rr == nil { 314 314 + results <- bidResult{} 315 315 + return 316 316 + } 317 317 + if !rr.GetAccepted() { 318 318 + if rr.GetRejectClass() == millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE { 319 319 + results <- bidResult{sess: sess, rank: rank, incompatible: true, reason: rr.GetRejectReason()} 317 320 return 318 321 } 319 319 - results <- bidResult{sess: sess, lease: lease, score: rr.GetScore()} 320 320 - }(sess) 322 322 + results <- bidResult{} 323 323 + return 324 324 + } 325 325 + results <- bidResult{sess: sess, lease: lease, rank: rank} 326 326 + } 327 327 + window := m.cfg.TopK 328 328 + if window <= 0 { 329 329 + window = len(candidates) 321 330 } 322 322 - wg.Wait() 323 323 - close(results) 331 331 + next := 0 332 332 + inFlight := 0 333 333 + for next < len(candidates) && inFlight < window { 334 334 + inFlight++ 335 335 + go ask(next, candidates[next]) 336 336 + next++ 337 337 + } 324 338 325 339 var winner *bidResult 326 340 var losers []*RemoteLease 327 327 - for r := range results { 328 328 - if winner == nil || r.score > winner.score { 341 341 + var incompatible []string 342 342 + softReject := false 343 343 + for inFlight > 0 { 344 344 + r := <-results 345 345 + inFlight-- 346 346 + if r.incompatible { 347 347 + if r.reason != "" { 348 348 + incompatible = append(incompatible, r.reason) 349 349 + } 350 350 + } else if r.lease == nil { 351 351 + softReject = true 352 352 + } 353 353 + if r.lease == nil { 354 354 + for winner == nil && next < len(candidates) && inFlight < window { 355 355 + inFlight++ 356 356 + go ask(next, candidates[next]) 357 357 + next++ 358 358 + } 359 359 + continue 360 360 + } 361 361 + if winner == nil || r.rank < winner.rank { 329 362 if winner != nil { 330 363 losers = append(losers, winner.lease) 331 364 } ··· 341 374 } 342 375 343 376 if winner == nil { 377 377 + if len(incompatible) > 0 && !softReject { 378 378 + return nil, fmt.Errorf("no compatible executor for %s: %s", engineName, strings.Join(incompatible, "; ")) 379 379 + } 344 380 return nil, nil 345 381 } 346 382 return winner.lease, nil 347 383 } 348 384 349 349 - func (m *Mill) rankCandidates(engineName string) []*millSession { 385 385 + func (m *Mill) rankCandidates(engineName string, requiredLabels []string) []*millSession { 350 386 m.mu.Lock() 351 387 defer m.mu.Unlock() 352 388 353 389 type ranked struct { 354 390 sess *millSession 355 355 - seats uint32 391 391 + worst float64 392 392 + sum float64 356 393 } 357 394 var rs []ranked 358 395 for _, s := range m.sessions { ··· 362 399 if s.snapshot == nil { 363 400 continue 364 401 } 365 365 - es, ok := s.snapshot.GetEngines()[engineName] 366 366 - if !ok || es.GetFreeSeats() == 0 { 402 402 + ea, ok := s.snapshot.GetEngines()[engineName] 403 403 + if !ok || !ea.GetAvailable() { 367 404 continue 368 405 } 369 369 - rs = append(rs, ranked{sess: s, seats: es.GetFreeSeats()}) 406 406 + if !hasLabels(s.labels, requiredLabels) { 407 407 + continue 408 408 + } 409 409 + worst, sum := loadScore(ea.GetLoad()) 410 410 + rs = append(rs, ranked{sess: s, worst: worst, sum: sum}) 370 411 } 371 371 - sort.SliceStable(rs, func(i, j int) bool { return rs[i].seats > rs[j].seats }) 412 412 + slices.SortStableFunc(rs, func(a, b ranked) int { 413 413 + if a.worst < b.worst { 414 414 + return -1 415 415 + } 416 416 + if a.worst > b.worst { 417 417 + return 1 418 418 + } 419 419 + if a.sum < b.sum { 420 420 + return -1 421 421 + } 422 422 + if a.sum > b.sum { 423 423 + return 1 424 424 + } 425 425 + return 0 426 426 + }) 372 427 373 428 out := make([]*millSession, len(rs)) 374 429 for i := range rs { ··· 377 432 return out 378 433 } 379 434 380 380 - // --- commit + wait (RunStep) ---------------------------------------------- 435 435 + func loadScore(load map[string]float64) (worst, sum float64) { 436 436 + for _, v := range load { 437 437 + if v > worst { 438 438 + worst = v 439 439 + } 440 440 + sum += v 441 441 + } 442 442 + return worst, sum 443 443 + } 444 444 + 445 445 + func requiredLabels(wf *models.Workflow) []string { 446 446 + st, ok := wf.Data.(*millWorkflowState) 447 447 + if !ok || st == nil { 448 448 + return nil 449 449 + } 450 450 + return st.RawWorkflow.RunsOn 451 451 + } 452 452 + 453 453 + func hasLabels(labels []string, required []string) bool { 454 454 + for _, want := range required { 455 455 + if !slices.Contains(labels, want) { 456 456 + return false 457 457 + } 458 458 + } 459 459 + return true 460 460 + } 381 461 382 462 func (m *Mill) commitAndWait(ctx context.Context, wf *models.Workflow, unlocked []secrets.UnlockedSecret) error { 383 463 st, ok := wf.Data.(*millWorkflowState)

+320

spindle/mill/mill_test.go

Reviewed

··· 5 5 "errors" 6 6 "io" 7 7 "log/slog" 8 8 + "strings" 8 9 "testing" 9 10 "time" 10 11 ··· 33 34 } 34 35 } 35 36 37 37 + func testWorkflowWithRunsOn(name string, runsOn []string) *models.Workflow { 38 38 + wf := testWorkflow(name) 39 39 + wf.Data.(*millWorkflowState).RawWorkflow.RunsOn = runsOn 40 40 + return wf 41 41 + } 42 42 + 43 43 + func addCandidateSession(t *testing.T, m *Mill, nodeID string, labels []string, load float64, enc messageEncoder) *millSession { 44 44 + t.Helper() 45 45 + if enc == nil { 46 46 + enc = scriptedEncoder(func(*millproto.Message) error { return nil }) 47 47 + } 48 48 + sess := newSession(nodeID, enc, slog.New(slog.NewTextHandler(io.Discard, nil))) 49 49 + sess.labels = labels 50 50 + sess.snapshot = &millv1.NodeSnapshot{ 51 51 + NodeId: nodeID, 52 52 + Engines: map[string]*millv1.EngineAvailability{ 53 53 + "dummy": {Available: load < 1.0, Load: map[string]float64{"slots": load}}, 54 54 + }, 55 55 + } 56 56 + m.mu.Lock() 57 57 + m.sessions[nodeID] = sess 58 58 + m.mu.Unlock() 59 59 + return sess 60 60 + } 61 61 + 62 62 + func assertRankedNodes(t *testing.T, got []*millSession, want []string) { 63 63 + t.Helper() 64 64 + if len(got) != len(want) { 65 65 + t.Fatalf("rankCandidates() returned %d candidates, want %d: got %v want %v", len(got), len(want), sessionIDs(got), want) 66 66 + } 67 67 + for i := range want { 68 68 + if got[i].nodeID != want[i] { 69 69 + t.Fatalf("rankCandidates()[%d] = %q, want %q; full order got %v want %v", i, got[i].nodeID, want[i], sessionIDs(got), want) 70 70 + } 71 71 + } 72 72 + } 73 73 + 74 74 + func sessionIDs(sessions []*millSession) []string { 75 75 + out := make([]string, len(sessions)) 76 76 + for i, sess := range sessions { 77 77 + out[i] = sess.nodeID 78 78 + } 79 79 + return out 80 80 + } 81 81 + 82 82 + func sameStringMultiset(a, b []string) bool { 83 83 + if len(a) != len(b) { 84 84 + return false 85 85 + } 86 86 + counts := make(map[string]int, len(a)) 87 87 + for _, s := range a { 88 88 + counts[s]++ 89 89 + } 90 90 + for _, s := range b { 91 91 + if counts[s] == 0 { 92 92 + return false 93 93 + } 94 94 + counts[s]-- 95 95 + } 96 96 + return true 97 97 + } 98 98 + 99 99 + type reserveReply struct { 100 100 + accepted bool 101 101 + rejectClass millv1.RejectClass 102 102 + reason string 103 103 + } 104 104 + 105 105 + func addReplyingCandidateSession(t *testing.T, m *Mill, nodeID string, labels []string, load float64, asked chan<- string, reply reserveReply) *millSession { 106 106 + t.Helper() 107 107 + var sess *millSession 108 108 + sess = addCandidateSession(t, m, nodeID, labels, load, scriptedEncoder(func(msg *millproto.Message) error { 109 109 + rs := msg.GetReserveSeat() 110 110 + if rs == nil { 111 111 + return nil 112 112 + } 113 113 + if asked != nil { 114 114 + asked <- nodeID 115 115 + } 116 116 + sess.deliver(rs.GetLeaseId(), &millproto.Message{ReserveResult: &millv1.ReserveResult{ 117 117 + LeaseId: rs.GetLeaseId(), 118 118 + Accepted: reply.accepted, 119 119 + RejectReason: reply.reason, 120 120 + RejectClass: reply.rejectClass, 121 121 + }}) 122 122 + return nil 123 123 + })) 124 124 + return sess 125 125 + } 126 126 + 127 127 + func drainAsked(ch <-chan string) []string { 128 128 + var out []string 129 129 + for { 130 130 + select { 131 131 + case nodeID := <-ch: 132 132 + out = append(out, nodeID) 133 133 + default: 134 134 + return out 135 135 + } 136 136 + } 137 137 + } 138 138 + 36 139 func TestCommitRetriesAfterSessionCloseBeforeCommitted(t *testing.T) { 37 140 l := slog.New(slog.NewTextHandler(io.Discard, nil)) 38 141 m := New(l, Config{BidTimeout: 25 * time.Millisecond, ReconnectGrace: time.Second}) ··· 138 241 } 139 242 } 140 243 244 244 + func TestRankCandidatesFiltersRequiredLabelsWithANDSemantics(t *testing.T) { 245 245 + m := New(slog.New(slog.NewTextHandler(io.Discard, nil)), Config{}) 246 246 + addCandidateSession(t, m, "linux-high", []string{"linux"}, 0.0, nil) 247 247 + addCandidateSession(t, m, "linux-arm", []string{"linux", "arm64"}, 0.25, nil) 248 248 + addCandidateSession(t, m, "unlabeled", nil, 0.5, nil) 249 249 + addCandidateSession(t, m, "linux-arm-gpu", []string{"linux", "arm64", "gpu"}, 0.75, nil) 250 250 + addCandidateSession(t, m, "linux-arm-full", []string{"linux", "arm64"}, 1.0, nil) 251 251 + 252 252 + tests := []struct { 253 253 + name string 254 254 + requiredLabels []string 255 255 + want []string 256 256 + }{ 257 257 + { 258 258 + name: "no required labels keeps old capacity ranking", 259 259 + want: []string{"linux-high", "linux-arm", "unlabeled", "linux-arm-gpu"}, 260 260 + }, 261 261 + { 262 262 + name: "single required label includes every candidate carrying it", 263 263 + requiredLabels: []string{"linux"}, 264 264 + want: []string{"linux-high", "linux-arm", "linux-arm-gpu"}, 265 265 + }, 266 266 + { 267 267 + name: "all required labels must be present", 268 268 + requiredLabels: []string{"linux", "arm64"}, 269 269 + want: []string{"linux-arm", "linux-arm-gpu"}, 270 270 + }, 271 271 + { 272 272 + name: "one missing required label excludes the candidate", 273 273 + requiredLabels: []string{"linux", "arm64", "gpu"}, 274 274 + want: []string{"linux-arm-gpu"}, 275 275 + }, 276 276 + { 277 277 + name: "unknown required label leaves no candidate", 278 278 + requiredLabels: []string{"linux", "arm64", "metal"}, 279 279 + }, 280 280 + } 281 281 + 282 282 + for _, tt := range tests { 283 283 + t.Run(tt.name, func(t *testing.T) { 284 284 + assertRankedNodes(t, m.rankCandidates("dummy", tt.requiredLabels), tt.want) 285 285 + }) 286 286 + } 287 287 + } 288 288 + 289 289 + func TestPlaceWithMissingRequiredLabelsStaysPendingWithoutReserve(t *testing.T) { 290 290 + l := slog.New(slog.NewTextHandler(io.Discard, nil)) 291 291 + m := New(l, Config{BidTimeout: 10 * time.Millisecond}) 292 292 + reserveSent := make(chan struct{}, 1) 293 293 + addCandidateSession(t, m, "linux-only", []string{"linux"}, 0.75, scriptedEncoder(func(msg *millproto.Message) error { 294 294 + if msg.GetReserveSeat() != nil { 295 295 + select { 296 296 + case reserveSent <- struct{}{}: 297 297 + default: 298 298 + } 299 299 + } 300 300 + return nil 301 301 + })) 302 302 + wf := testWorkflowWithRunsOn("build", []string{"linux", "arm64"}) 303 303 + wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"} 304 304 + 305 305 + ctx, cancel := context.WithTimeout(context.Background(), 120*time.Millisecond) 306 306 + defer cancel() 307 307 + _, err := m.place(ctx, "dummy", wid, wf) 308 308 + if err != context.DeadlineExceeded { 309 309 + t.Fatalf("place() error = %v, want DeadlineExceeded while job remains pending", err) 310 310 + } 311 311 + select { 312 312 + case <-reserveSent: 313 313 + t.Fatal("place() sent ReserveSeat to executor missing a required label") 314 314 + default: 315 315 + } 316 316 + } 317 317 + 318 318 + func TestPlaceKeepsAskingBelowTopKAfterIncompatibleRejects(t *testing.T) { 319 319 + l := slog.New(slog.NewTextHandler(io.Discard, nil)) 320 320 + m := New(l, Config{TopK: 2, BidTimeout: time.Second}) 321 321 + asked := make(chan string, 4) 322 322 + addReplyingCandidateSession(t, m, "wrong-label", []string{"linux"}, 0.0, asked, reserveReply{ 323 323 + accepted: false, 324 324 + rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE, 325 325 + reason: "wrong-label should not be asked", 326 326 + }) 327 327 + addReplyingCandidateSession(t, m, "incompatible-a", []string{"linux", "arm64"}, 0.0, asked, reserveReply{ 328 328 + accepted: false, 329 329 + rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE, 330 330 + reason: "no runner", 331 331 + }) 332 332 + addReplyingCandidateSession(t, m, "incompatible-b", []string{"linux", "arm64"}, 0.25, asked, reserveReply{ 333 333 + accepted: false, 334 334 + rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE, 335 335 + reason: "bad image", 336 336 + }) 337 337 + addReplyingCandidateSession(t, m, "compatible-below-window", []string{"linux", "arm64"}, 0.5, asked, reserveReply{ 338 338 + accepted: true, 339 339 + }) 340 340 + 341 341 + wf := testWorkflowWithRunsOn("build", []string{"linux", "arm64"}) 342 342 + wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"} 343 343 + ctx, cancel := context.WithTimeout(context.Background(), time.Second) 344 344 + defer cancel() 345 345 + slot, err := m.place(ctx, "dummy", wid, wf) 346 346 + if err != nil { 347 347 + t.Fatalf("place() error = %v, want compatible lower-ranked executor", err) 348 348 + } 349 349 + defer slot.Release() 350 350 + 351 351 + ms, ok := slot.(*millSlot) 352 352 + if !ok { 353 353 + t.Fatalf("place() slot type = %T, want *millSlot", slot) 354 354 + } 355 355 + if ms.lease.nodeID != "compatible-below-window" { 356 356 + t.Fatalf("place() chose node %q, want compatible-below-window", ms.lease.nodeID) 357 357 + } 358 358 + if got, want := drainAsked(asked), []string{"incompatible-a", "incompatible-b", "compatible-below-window"}; !sameStringMultiset(got, want) { 359 359 + t.Fatalf("ReserveSeat asked nodes = %v, want %v", got, want) 360 360 + } 361 361 + } 362 362 + 363 363 + func TestPlaceKeepsTransientOnlyRejectsPending(t *testing.T) { 364 364 + l := slog.New(slog.NewTextHandler(io.Discard, nil)) 365 365 + m := New(l, Config{TopK: 1, BidTimeout: 10 * time.Millisecond}) 366 366 + asked := make(chan string, 2) 367 367 + addReplyingCandidateSession(t, m, "busy-a", []string{"linux", "arm64"}, 0.5, asked, reserveReply{ 368 368 + accepted: false, 369 369 + rejectClass: millv1.RejectClass_REJECT_CLASS_TRANSIENT, 370 370 + reason: "draining", 371 371 + }) 372 372 + addReplyingCandidateSession(t, m, "busy-b", []string{"linux", "arm64"}, 0.75, asked, reserveReply{ 373 373 + accepted: false, 374 374 + rejectClass: millv1.RejectClass_REJECT_CLASS_TRANSIENT, 375 375 + reason: "no slot", 376 376 + }) 377 377 + 378 378 + wf := testWorkflowWithRunsOn("build", []string{"linux", "arm64"}) 379 379 + wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"} 380 380 + ctx, cancel := context.WithTimeout(context.Background(), 120*time.Millisecond) 381 381 + defer cancel() 382 382 + _, err := m.place(ctx, "dummy", wid, wf) 383 383 + if err != context.DeadlineExceeded { 384 384 + t.Fatalf("place() error = %v, want DeadlineExceeded while transient rejects leave job pending", err) 385 385 + } 386 386 + if got, want := drainAsked(asked), []string{"busy-a", "busy-b"}; !sameStringMultiset(got, want) { 387 387 + t.Fatalf("ReserveSeat asked nodes = %v, want %v", got, want) 388 388 + } 389 389 + } 390 390 + 391 391 + func TestPlaceReportsOnlyEligiblePermanentIncompatibleRejects(t *testing.T) { 392 392 + l := slog.New(slog.NewTextHandler(io.Discard, nil)) 393 393 + m := New(l, Config{TopK: 1, BidTimeout: time.Second}) 394 394 + asked := make(chan string, 3) 395 395 + addReplyingCandidateSession(t, m, "wrong-label", []string{"linux"}, 0.25, asked, reserveReply{ 396 396 + accepted: false, 397 397 + rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE, 398 398 + reason: "wrong-label should not be asked", 399 399 + }) 400 400 + addReplyingCandidateSession(t, m, "incompatible-a", []string{"linux", "arm64"}, 0.5, asked, reserveReply{ 401 401 + accepted: false, 402 402 + rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE, 403 403 + reason: "no qemu-system-aarch64", 404 404 + }) 405 405 + addReplyingCandidateSession(t, m, "incompatible-b", []string{"linux", "arm64"}, 0.75, asked, reserveReply{ 406 406 + accepted: false, 407 407 + rejectClass: millv1.RejectClass_REJECT_CLASS_INCOMPATIBLE, 408 408 + reason: "image arch unsupported", 409 409 + }) 410 410 + 411 411 + wf := testWorkflowWithRunsOn("build", []string{"linux", "arm64"}) 412 412 + wid := models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"} 413 413 + ctx, cancel := context.WithTimeout(context.Background(), time.Second) 414 414 + defer cancel() 415 415 + _, err := m.place(ctx, "dummy", wid, wf) 416 416 + if err == nil { 417 417 + t.Fatal("place() error = nil, want permanent incompatible failure") 418 418 + } 419 419 + errText := err.Error() 420 420 + for _, want := range []string{"no compatible executor for dummy", "no qemu-system-aarch64", "image arch unsupported"} { 421 421 + if !strings.Contains(errText, want) { 422 422 + t.Fatalf("place() error = %q, want it to contain %q", errText, want) 423 423 + } 424 424 + } 425 425 + if strings.Contains(errText, "wrong-label should not be asked") { 426 426 + t.Fatalf("place() error = %q, included a missing-label candidate as incompatible", errText) 427 427 + } 428 428 + if got, want := drainAsked(asked), []string{"incompatible-a", "incompatible-b"}; !sameStringMultiset(got, want) { 429 429 + t.Fatalf("ReserveSeat asked nodes = %v, want %v", got, want) 430 430 + } 431 431 + } 432 432 + 141 433 func TestMaxPendingRejects(t *testing.T) { 142 434 l := slog.New(slog.NewTextHandler(io.Discard, nil)) 143 435 m := New(l, Config{MaxPending: 1}) ··· 175 467 t.Fatalf("request left %d pending waiters, want 0", pending) 176 468 } 177 469 } 470 470 + 471 471 + func TestFallbackBidGetsFullTimeout(t *testing.T) { 472 472 + m := New(discardLogger(), Config{TopK: 1, BidTimeout: 30 * time.Millisecond}) 473 473 + addCandidateSession(t, m, "silent", nil, 0, nil) 474 474 + var fallback *millSession 475 475 + fallback = addCandidateSession(t, m, "fallback", nil, 0.5, scriptedEncoder(func(msg *millproto.Message) error { 476 476 + if reserve := msg.GetReserveSeat(); reserve != nil { 477 477 + leaseID := reserve.GetLeaseId() 478 478 + time.AfterFunc(10*time.Millisecond, func() { 479 479 + fallback.deliver(leaseID, &millproto.Message{ReserveResult: &millv1.ReserveResult{ 480 480 + LeaseId: leaseID, 481 481 + Accepted: true, 482 482 + }}) 483 483 + }) 484 484 + } 485 485 + return nil 486 486 + })) 487 487 + 488 488 + ctx, cancel := context.WithTimeout(context.Background(), 200*time.Millisecond) 489 489 + defer cancel() 490 490 + lease, err := m.bid(ctx, "dummy", testWorkflow("build")) 491 491 + if err != nil { 492 492 + t.Fatalf("bid: %v", err) 493 493 + } 494 494 + if lease == nil || lease.nodeID != "fallback" { 495 495 + t.Fatalf("bid winner = %+v, want fallback after silent incumbent times out", lease) 496 496 + } 497 497 + }

+163 -112

spindle/mill/proto/gen/mill.pb.go

Reviewed

··· 22 22 _ = protoimpl.EnforceVersion(protoimpl.MaxVersion - 20) 23 23 ) 24 24 25 25 + type RejectClass int32 26 26 + 27 27 + const ( 28 28 + RejectClass_REJECT_CLASS_UNSPECIFIED RejectClass = 0 29 29 + RejectClass_REJECT_CLASS_TRANSIENT RejectClass = 1 30 30 + RejectClass_REJECT_CLASS_INCOMPATIBLE RejectClass = 2 31 31 + ) 32 32 + 33 33 + // Enum value maps for RejectClass. 34 34 + var ( 35 35 + RejectClass_name = map[int32]string{ 36 36 + 0: "REJECT_CLASS_UNSPECIFIED", 37 37 + 1: "REJECT_CLASS_TRANSIENT", 38 38 + 2: "REJECT_CLASS_INCOMPATIBLE", 39 39 + } 40 40 + RejectClass_value = map[string]int32{ 41 41 + "REJECT_CLASS_UNSPECIFIED": 0, 42 42 + "REJECT_CLASS_TRANSIENT": 1, 43 43 + "REJECT_CLASS_INCOMPATIBLE": 2, 44 44 + } 45 45 + ) 46 46 + 47 47 + func (x RejectClass) Enum() *RejectClass { 48 48 + p := new(RejectClass) 49 49 + *p = x 50 50 + return p 51 51 + } 52 52 + 53 53 + func (x RejectClass) String() string { 54 54 + return protoimpl.X.EnumStringOf(x.Descriptor(), protoreflect.EnumNumber(x)) 55 55 + } 56 56 + 57 57 + func (RejectClass) Descriptor() protoreflect.EnumDescriptor { 58 58 + return file_spindle_mill_v1_mill_proto_enumTypes[0].Descriptor() 59 59 + } 60 60 + 61 61 + func (RejectClass) Type() protoreflect.EnumType { 62 62 + return &file_spindle_mill_v1_mill_proto_enumTypes[0] 63 63 + } 64 64 + 65 65 + func (x RejectClass) Number() protoreflect.EnumNumber { 66 66 + return protoreflect.EnumNumber(x) 67 67 + } 68 68 + 69 69 + // Deprecated: Use RejectClass.Descriptor instead. 70 70 + func (RejectClass) EnumDescriptor() ([]byte, []int) { 71 71 + return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{0} 72 72 + } 73 73 + 25 74 // Hello is the first frame an executor sends after dialing the mill. It 26 75 // carries the static traits of the node plus a resume hint. 27 76 type Hello struct { ··· 32 81 NodeId string `protobuf:"bytes,2,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"` 33 82 // engine names this node can run ("microvm", "nixery"). 34 83 Engines []string `protobuf:"bytes,3,rep,name=engines,proto3" json:"engines,omitempty"` 35 35 - // GOARCH of the node, so the mill won't place arch-incompatible jobs. 84 84 + // GOARCH of the node, retained as an informational trait for logs. 36 85 Arch string `protobuf:"bytes,4,opt,name=arch,proto3" json:"arch,omitempty"` 37 86 // the highest relay offset the executor believes it has sent; a resume hint. 38 38 - LastOffset uint64 `protobuf:"varint,5,opt,name=last_offset,json=lastOffset,proto3" json:"last_offset,omitempty"` 87 87 + LastOffset uint64 `protobuf:"varint,5,opt,name=last_offset,json=lastOffset,proto3" json:"last_offset,omitempty"` 88 88 + // opaque operator-defined labels used for runs_on matching. 89 89 + Labels []string `protobuf:"bytes,6,rep,name=labels,proto3" json:"labels,omitempty"` 39 90 unknownFields protoimpl.UnknownFields 40 91 sizeCache protoimpl.SizeCache 41 92 } ··· 105 156 return 0 106 157 } 107 158 159 159 + func (x *Hello) GetLabels() []string { 160 160 + if x != nil { 161 161 + return x.Labels 162 162 + } 163 163 + return nil 164 164 + } 165 165 + 108 166 // Resume is the mill's reply to Hello. The executor replays every buffered 109 167 // relay entry with offset strictly greater than ack_offset before sending new 110 168 // ones. ··· 152 210 return 0 153 211 } 154 212 155 155 - // EngineSnapshot is the changing per-engine state on a node. free_seats is a 156 156 - // coarse "can you take more" hint; 0 also means draining (the mill treats a 157 157 - // draining node as a full one until it leaves). The resource fields are coarse 158 158 - // budget headroom for ranking only; the executor makes the real yes/no call in 159 159 - // ReserveResult. 160 160 - type EngineSnapshot struct { 161 161 - state protoimpl.MessageState `protogen:"open.v1"` 162 162 - FreeSeats uint32 `protobuf:"varint,1,opt,name=free_seats,json=freeSeats,proto3" json:"free_seats,omitempty"` 163 163 - FreeMemoryMib int64 `protobuf:"varint,2,opt,name=free_memory_mib,json=freeMemoryMib,proto3" json:"free_memory_mib,omitempty"` 164 164 - FreeVcpus int64 `protobuf:"varint,3,opt,name=free_vcpus,json=freeVcpus,proto3" json:"free_vcpus,omitempty"` 165 165 - FreeDiskMib int64 `protobuf:"varint,4,opt,name=free_disk_mib,json=freeDiskMib,proto3" json:"free_disk_mib,omitempty"` 213 213 + // EngineAvailability is the changing per-engine state on a node. The executor 214 214 + // reports opaque load metrics (higher = more loaded); the mill ranks by the 215 215 + // worst load and uses the sum as a tie-breaker. The mill does not know what 216 216 + // the metric keys mean. 217 217 + type EngineAvailability struct { 218 218 + state protoimpl.MessageState `protogen:"open.v1"` 219 219 + Available bool `protobuf:"varint,1,opt,name=available,proto3" json:"available,omitempty"` 220 220 + // engine-defined load metrics. keys are opaque to the mill. 221 221 + Load map[string]float64 `protobuf:"bytes,2,rep,name=load,proto3" json:"load,omitempty" protobuf_key:"bytes,1,opt,name=key" protobuf_val:"fixed64,2,opt,name=value"` 166 222 unknownFields protoimpl.UnknownFields 167 223 sizeCache protoimpl.SizeCache 168 224 } 169 225 170 170 - func (x *EngineSnapshot) Reset() { 171 171 - *x = EngineSnapshot{} 226 226 + func (x *EngineAvailability) Reset() { 227 227 + *x = EngineAvailability{} 172 228 mi := &file_spindle_mill_v1_mill_proto_msgTypes[2] 173 229 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) 174 230 ms.StoreMessageInfo(mi) 175 231 } 176 232 177 177 - func (x *EngineSnapshot) String() string { 233 233 + func (x *EngineAvailability) String() string { 178 234 return protoimpl.X.MessageStringOf(x) 179 235 } 180 236 181 181 - func (*EngineSnapshot) ProtoMessage() {} 237 237 + func (*EngineAvailability) ProtoMessage() {} 182 238 183 183 - func (x *EngineSnapshot) ProtoReflect() protoreflect.Message { 239 239 + func (x *EngineAvailability) ProtoReflect() protoreflect.Message { 184 240 mi := &file_spindle_mill_v1_mill_proto_msgTypes[2] 185 241 if x != nil { 186 242 ms := protoimpl.X.MessageStateOf(protoimpl.Pointer(x)) ··· 192 248 return mi.MessageOf(x) 193 249 } 194 250 195 195 - // Deprecated: Use EngineSnapshot.ProtoReflect.Descriptor instead. 196 196 - func (*EngineSnapshot) Descriptor() ([]byte, []int) { 251 251 + // Deprecated: Use EngineAvailability.ProtoReflect.Descriptor instead. 252 252 + func (*EngineAvailability) Descriptor() ([]byte, []int) { 197 253 return file_spindle_mill_v1_mill_proto_rawDescGZIP(), []int{2} 198 254 } 199 255 200 200 - func (x *EngineSnapshot) GetFreeSeats() uint32 { 201 201 - if x != nil { 202 202 - return x.FreeSeats 203 203 - } 204 204 - return 0 205 205 - } 206 206 - 207 207 - func (x *EngineSnapshot) GetFreeMemoryMib() int64 { 208 208 - if x != nil { 209 209 - return x.FreeMemoryMib 210 210 - } 211 211 - return 0 212 212 - } 213 213 - 214 214 - func (x *EngineSnapshot) GetFreeVcpus() int64 { 256 256 + func (x *EngineAvailability) GetAvailable() bool { 215 257 if x != nil { 216 216 - return x.FreeVcpus 258 258 + return x.Available 217 259 } 218 218 - return 0 260 260 + return false 219 261 } 220 262 221 221 - func (x *EngineSnapshot) GetFreeDiskMib() int64 { 263 263 + func (x *EngineAvailability) GetLoad() map[string]float64 { 222 264 if x != nil { 223 223 - return x.FreeDiskMib 265 265 + return x.Load 224 266 } 225 225 - return 0 267 267 + return nil 226 268 } 227 269 228 270 // NodeSnapshot is pushed on connect, periodically, and right after any state 229 271 // change (reserve, commit, terminal). 230 272 type NodeSnapshot struct { 231 231 - state protoimpl.MessageState `protogen:"open.v1"` 232 232 - NodeId string `protobuf:"bytes,1,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"` 233 233 - Seq uint64 `protobuf:"varint,2,opt,name=seq,proto3" json:"seq,omitempty"` 234 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"` 273 273 + state protoimpl.MessageState `protogen:"open.v1"` 274 274 + NodeId string `protobuf:"bytes,1,opt,name=node_id,json=nodeId,proto3" json:"node_id,omitempty"` 275 275 + Seq uint64 `protobuf:"varint,2,opt,name=seq,proto3" json:"seq,omitempty"` 276 276 + Engines map[string]*EngineAvailability `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 277 unknownFields protoimpl.UnknownFields 236 278 sizeCache protoimpl.SizeCache 237 279 } ··· 280 322 return 0 281 323 } 282 324 283 283 - func (x *NodeSnapshot) GetEngines() map[string]*EngineSnapshot { 325 325 + func (x *NodeSnapshot) GetEngines() map[string]*EngineAvailability { 284 326 if x != nil { 285 327 return x.Engines 286 328 } ··· 387 429 388 430 // ReserveResult is the executor's accept/reject for a ReserveSeat. 389 431 type ReserveResult struct { 390 390 - state protoimpl.MessageState `protogen:"open.v1"` 391 391 - LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"` 392 392 - Accepted bool `protobuf:"varint,2,opt,name=accepted,proto3" json:"accepted,omitempty"` 393 393 - RejectReason string `protobuf:"bytes,3,opt,name=reject_reason,json=rejectReason,proto3" json:"reject_reason,omitempty"` 394 394 - // optional bid score the mill ranks accepted leases by (higher is better). 395 395 - Score float64 `protobuf:"fixed64,4,opt,name=score,proto3" json:"score,omitempty"` 432 432 + state protoimpl.MessageState `protogen:"open.v1"` 433 433 + LeaseId string `protobuf:"bytes,1,opt,name=lease_id,json=leaseId,proto3" json:"lease_id,omitempty"` 434 434 + Accepted bool `protobuf:"varint,2,opt,name=accepted,proto3" json:"accepted,omitempty"` 435 435 + RejectReason string `protobuf:"bytes,3,opt,name=reject_reason,json=rejectReason,proto3" json:"reject_reason,omitempty"` 436 436 + RejectClass RejectClass `protobuf:"varint,4,opt,name=reject_class,json=rejectClass,proto3,enum=spindle.mill.v1.RejectClass" json:"reject_class,omitempty"` 396 437 unknownFields protoimpl.UnknownFields 397 438 sizeCache protoimpl.SizeCache 398 439 } ··· 448 489 return "" 449 490 } 450 491 451 451 - func (x *ReserveResult) GetScore() float64 { 492 492 + func (x *ReserveResult) GetRejectClass() RejectClass { 452 493 if x != nil { 453 453 - return x.Score 494 494 + return x.RejectClass 454 495 } 455 455 - return 0 496 496 + return RejectClass_REJECT_CLASS_UNSPECIFIED 456 497 } 457 498 458 499 // Secret is a single unlocked secret, sent only inside CommitLease. ··· 1134 1175 1135 1176 const file_spindle_mill_v1_mill_proto_rawDesc = "" + 1136 1177 "\n" + 1137 1137 - "\x1aspindle/mill/v1/mill.proto\x12\x0fspindle.mill.v1\x1a\x1bbuf/validate/validate.proto\"\xa3\x01\n" + 1178 1178 + "\x1aspindle/mill/v1/mill.proto\x12\x0fspindle.mill.v1\x1a\x1bbuf/validate/validate.proto\"\xbb\x01\n" + 1138 1179 "\x05Hello\x12)\n" + 1139 1180 "\x10protocol_version\x18\x01 \x01(\rR\x0fprotocolVersion\x12 \n" + 1140 1181 "\anode_id\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\x06nodeId\x12\x18\n" + 1141 1182 "\aengines\x18\x03 \x03(\tR\aengines\x12\x12\n" + 1142 1183 "\x04arch\x18\x04 \x01(\tR\x04arch\x12\x1f\n" + 1143 1184 "\vlast_offset\x18\x05 \x01(\x04R\n" + 1144 1144 - "lastOffset\"'\n" + 1185 1185 + "lastOffset\x12\x16\n" + 1186 1186 + "\x06labels\x18\x06 \x03(\tR\x06labels\"'\n" + 1145 1187 "\x06Resume\x12\x1d\n" + 1146 1188 "\n" + 1147 1147 - "ack_offset\x18\x01 \x01(\x04R\tackOffset\"\x9a\x01\n" + 1148 1148 - "\x0eEngineSnapshot\x12\x1d\n" + 1149 1149 - "\n" + 1150 1150 - "free_seats\x18\x01 \x01(\rR\tfreeSeats\x12&\n" + 1151 1151 - "\x0ffree_memory_mib\x18\x02 \x01(\x03R\rfreeMemoryMib\x12\x1d\n" + 1152 1152 - "\n" + 1153 1153 - "free_vcpus\x18\x03 \x01(\x03R\tfreeVcpus\x12\"\n" + 1154 1154 - "\rfree_disk_mib\x18\x04 \x01(\x03R\vfreeDiskMib\"\xdc\x01\n" + 1189 1189 + "ack_offset\x18\x01 \x01(\x04R\tackOffset\"\xae\x01\n" + 1190 1190 + "\x12EngineAvailability\x12\x1c\n" + 1191 1191 + "\tavailable\x18\x01 \x01(\bR\tavailable\x12A\n" + 1192 1192 + "\x04load\x18\x02 \x03(\v2-.spindle.mill.v1.EngineAvailability.LoadEntryR\x04load\x1a7\n" + 1193 1193 + "\tLoadEntry\x12\x10\n" + 1194 1194 + "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + 1195 1195 + "\x05value\x18\x02 \x01(\x01R\x05value:\x028\x01\"\xe0\x01\n" + 1155 1196 "\fNodeSnapshot\x12\x17\n" + 1156 1197 "\anode_id\x18\x01 \x01(\tR\x06nodeId\x12\x10\n" + 1157 1198 "\x03seq\x18\x02 \x01(\x04R\x03seq\x12D\n" + 1158 1158 - "\aengines\x18\x03 \x03(\v2*.spindle.mill.v1.NodeSnapshot.EnginesEntryR\aengines\x1a[\n" + 1199 1199 + "\aengines\x18\x03 \x03(\v2*.spindle.mill.v1.NodeSnapshot.EnginesEntryR\aengines\x1a_\n" + 1159 1200 "\fEnginesEntry\x12\x10\n" + 1160 1160 - "\x03key\x18\x01 \x01(\tR\x03key\x125\n" + 1161 1161 - "\x05value\x18\x02 \x01(\v2\x1f.spindle.mill.v1.EngineSnapshotR\x05value:\x028\x01\"\x80\x02\n" + 1201 1201 + "\x03key\x18\x01 \x01(\tR\x03key\x129\n" + 1202 1202 + "\x05value\x18\x02 \x01(\v2#.spindle.mill.v1.EngineAvailabilityR\x05value:\x028\x01\"\x80\x02\n" + 1162 1203 "\vReserveSeat\x12\"\n" + 1163 1204 "\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12,\n" + 1164 1205 "\rtarget_engine\x18\x02 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\ftargetEngine\x12*\n" + ··· 1167 1208 "\x04knot\x18\x05 \x01(\tR\x04knot\x12\x12\n" + 1168 1209 "\x04rkey\x18\x06 \x01(\tR\x04rkey\x12\x1f\n" + 1169 1210 "\vttl_seconds\x18\a \x01(\rR\n" + 1170 1170 - "ttlSeconds\"\x8a\x01\n" + 1211 1211 + "ttlSeconds\"\xb5\x01\n" + 1171 1212 "\rReserveResult\x12\"\n" + 1172 1213 "\blease_id\x18\x01 \x01(\tB\a\xbaH\x04r\x02\x10\x01R\aleaseId\x12\x1a\n" + 1173 1214 "\baccepted\x18\x02 \x01(\bR\baccepted\x12#\n" + 1174 1174 - "\rreject_reason\x18\x03 \x01(\tR\frejectReason\x12\x14\n" + 1175 1175 - "\x05score\x18\x04 \x01(\x01R\x05score\"0\n" + 1215 1215 + "\rreject_reason\x18\x03 \x01(\tR\frejectReason\x12?\n" + 1216 1216 + "\freject_class\x18\x04 \x01(\x0e2\x1c.spindle.mill.v1.RejectClassR\vrejectClass\"0\n" + 1176 1217 "\x06Secret\x12\x10\n" + 1177 1218 "\x03key\x18\x01 \x01(\tR\x03key\x12\x14\n" + 1178 1219 "\x05value\x18\x02 \x01(\tR\x05value\"d\n" + ··· 1234 1275 "\fstatus_event\n" + 1235 1276 "\blog_line\n" + 1236 1277 "\x0eattempt_result\n" + 1237 1237 - "\x03ack\x10\x01B0Z.tangled.org/core/spindle/mill/proto/gen;millv1b\x06proto3" 1278 1278 + "\x03ack\x10\x01*f\n" + 1279 1279 + "\vRejectClass\x12\x1c\n" + 1280 1280 + "\x18REJECT_CLASS_UNSPECIFIED\x10\x00\x12\x1a\n" + 1281 1281 + "\x16REJECT_CLASS_TRANSIENT\x10\x01\x12\x1d\n" + 1282 1282 + "\x19REJECT_CLASS_INCOMPATIBLE\x10\x02B0Z.tangled.org/core/spindle/mill/proto/gen;millv1b\x06proto3" 1238 1283 1239 1284 var ( 1240 1285 file_spindle_mill_v1_mill_proto_rawDescOnce sync.Once ··· 1248 1293 return file_spindle_mill_v1_mill_proto_rawDescData 1249 1294 } 1250 1295 1251 1251 - var file_spindle_mill_v1_mill_proto_msgTypes = make([]protoimpl.MessageInfo, 17) 1296 1296 + var file_spindle_mill_v1_mill_proto_enumTypes = make([]protoimpl.EnumInfo, 1) 1297 1297 + var file_spindle_mill_v1_mill_proto_msgTypes = make([]protoimpl.MessageInfo, 18) 1252 1298 var file_spindle_mill_v1_mill_proto_goTypes = []any{ 1253 1253 - (*Hello)(nil), // 0: spindle.mill.v1.Hello 1254 1254 - (*Resume)(nil), // 1: spindle.mill.v1.Resume 1255 1255 - (*EngineSnapshot)(nil), // 2: spindle.mill.v1.EngineSnapshot 1256 1256 - (*NodeSnapshot)(nil), // 3: spindle.mill.v1.NodeSnapshot 1257 1257 - (*ReserveSeat)(nil), // 4: spindle.mill.v1.ReserveSeat 1258 1258 - (*ReserveResult)(nil), // 5: spindle.mill.v1.ReserveResult 1259 1259 - (*Secret)(nil), // 6: spindle.mill.v1.Secret 1260 1260 - (*CommitLease)(nil), // 7: spindle.mill.v1.CommitLease 1261 1261 - (*Committed)(nil), // 8: spindle.mill.v1.Committed 1262 1262 - (*ReleaseLease)(nil), // 9: spindle.mill.v1.ReleaseLease 1263 1263 - (*CancelAttempt)(nil), // 10: spindle.mill.v1.CancelAttempt 1264 1264 - (*StatusEvent)(nil), // 11: spindle.mill.v1.StatusEvent 1265 1265 - (*LogLine)(nil), // 12: spindle.mill.v1.LogLine 1266 1266 - (*AttemptResult)(nil), // 13: spindle.mill.v1.AttemptResult 1267 1267 - (*Ack)(nil), // 14: spindle.mill.v1.Ack 1268 1268 - (*Message)(nil), // 15: spindle.mill.v1.Message 1269 1269 - nil, // 16: spindle.mill.v1.NodeSnapshot.EnginesEntry 1299 1299 + (RejectClass)(0), // 0: spindle.mill.v1.RejectClass 1300 1300 + (*Hello)(nil), // 1: spindle.mill.v1.Hello 1301 1301 + (*Resume)(nil), // 2: spindle.mill.v1.Resume 1302 1302 + (*EngineAvailability)(nil), // 3: spindle.mill.v1.EngineAvailability 1303 1303 + (*NodeSnapshot)(nil), // 4: spindle.mill.v1.NodeSnapshot 1304 1304 + (*ReserveSeat)(nil), // 5: spindle.mill.v1.ReserveSeat 1305 1305 + (*ReserveResult)(nil), // 6: spindle.mill.v1.ReserveResult 1306 1306 + (*Secret)(nil), // 7: spindle.mill.v1.Secret 1307 1307 + (*CommitLease)(nil), // 8: spindle.mill.v1.CommitLease 1308 1308 + (*Committed)(nil), // 9: spindle.mill.v1.Committed 1309 1309 + (*ReleaseLease)(nil), // 10: spindle.mill.v1.ReleaseLease 1310 1310 + (*CancelAttempt)(nil), // 11: spindle.mill.v1.CancelAttempt 1311 1311 + (*StatusEvent)(nil), // 12: spindle.mill.v1.StatusEvent 1312 1312 + (*LogLine)(nil), // 13: spindle.mill.v1.LogLine 1313 1313 + (*AttemptResult)(nil), // 14: spindle.mill.v1.AttemptResult 1314 1314 + (*Ack)(nil), // 15: spindle.mill.v1.Ack 1315 1315 + (*Message)(nil), // 16: spindle.mill.v1.Message 1316 1316 + nil, // 17: spindle.mill.v1.EngineAvailability.LoadEntry 1317 1317 + nil, // 18: spindle.mill.v1.NodeSnapshot.EnginesEntry 1270 1318 } 1271 1319 var file_spindle_mill_v1_mill_proto_depIdxs = []int32{ 1272 1272 - 16, // 0: spindle.mill.v1.NodeSnapshot.engines:type_name -> spindle.mill.v1.NodeSnapshot.EnginesEntry 1273 1273 - 6, // 1: spindle.mill.v1.CommitLease.secrets:type_name -> spindle.mill.v1.Secret 1274 1274 - 0, // 2: spindle.mill.v1.Message.hello:type_name -> spindle.mill.v1.Hello 1275 1275 - 1, // 3: spindle.mill.v1.Message.resume:type_name -> spindle.mill.v1.Resume 1276 1276 - 3, // 4: spindle.mill.v1.Message.node_snapshot:type_name -> spindle.mill.v1.NodeSnapshot 1277 1277 - 4, // 5: spindle.mill.v1.Message.reserve_seat:type_name -> spindle.mill.v1.ReserveSeat 1278 1278 - 5, // 6: spindle.mill.v1.Message.reserve_result:type_name -> spindle.mill.v1.ReserveResult 1279 1279 - 7, // 7: spindle.mill.v1.Message.commit_lease:type_name -> spindle.mill.v1.CommitLease 1280 1280 - 8, // 8: spindle.mill.v1.Message.committed:type_name -> spindle.mill.v1.Committed 1281 1281 - 9, // 9: spindle.mill.v1.Message.release_lease:type_name -> spindle.mill.v1.ReleaseLease 1282 1282 - 10, // 10: spindle.mill.v1.Message.cancel_attempt:type_name -> spindle.mill.v1.CancelAttempt 1283 1283 - 11, // 11: spindle.mill.v1.Message.status_event:type_name -> spindle.mill.v1.StatusEvent 1284 1284 - 12, // 12: spindle.mill.v1.Message.log_line:type_name -> spindle.mill.v1.LogLine 1285 1285 - 13, // 13: spindle.mill.v1.Message.attempt_result:type_name -> spindle.mill.v1.AttemptResult 1286 1286 - 14, // 14: spindle.mill.v1.Message.ack:type_name -> spindle.mill.v1.Ack 1287 1287 - 2, // 15: spindle.mill.v1.NodeSnapshot.EnginesEntry.value:type_name -> spindle.mill.v1.EngineSnapshot 1288 1288 - 16, // [16:16] is the sub-list for method output_type 1289 1289 - 16, // [16:16] is the sub-list for method input_type 1290 1290 - 16, // [16:16] is the sub-list for extension type_name 1291 1291 - 16, // [16:16] is the sub-list for extension extendee 1292 1292 - 0, // [0:16] is the sub-list for field type_name 1320 1320 + 17, // 0: spindle.mill.v1.EngineAvailability.load:type_name -> spindle.mill.v1.EngineAvailability.LoadEntry 1321 1321 + 18, // 1: spindle.mill.v1.NodeSnapshot.engines:type_name -> spindle.mill.v1.NodeSnapshot.EnginesEntry 1322 1322 + 0, // 2: spindle.mill.v1.ReserveResult.reject_class:type_name -> spindle.mill.v1.RejectClass 1323 1323 + 7, // 3: spindle.mill.v1.CommitLease.secrets:type_name -> spindle.mill.v1.Secret 1324 1324 + 1, // 4: spindle.mill.v1.Message.hello:type_name -> spindle.mill.v1.Hello 1325 1325 + 2, // 5: spindle.mill.v1.Message.resume:type_name -> spindle.mill.v1.Resume 1326 1326 + 4, // 6: spindle.mill.v1.Message.node_snapshot:type_name -> spindle.mill.v1.NodeSnapshot 1327 1327 + 5, // 7: spindle.mill.v1.Message.reserve_seat:type_name -> spindle.mill.v1.ReserveSeat 1328 1328 + 6, // 8: spindle.mill.v1.Message.reserve_result:type_name -> spindle.mill.v1.ReserveResult 1329 1329 + 8, // 9: spindle.mill.v1.Message.commit_lease:type_name -> spindle.mill.v1.CommitLease 1330 1330 + 9, // 10: spindle.mill.v1.Message.committed:type_name -> spindle.mill.v1.Committed 1331 1331 + 10, // 11: spindle.mill.v1.Message.release_lease:type_name -> spindle.mill.v1.ReleaseLease 1332 1332 + 11, // 12: spindle.mill.v1.Message.cancel_attempt:type_name -> spindle.mill.v1.CancelAttempt 1333 1333 + 12, // 13: spindle.mill.v1.Message.status_event:type_name -> spindle.mill.v1.StatusEvent 1334 1334 + 13, // 14: spindle.mill.v1.Message.log_line:type_name -> spindle.mill.v1.LogLine 1335 1335 + 14, // 15: spindle.mill.v1.Message.attempt_result:type_name -> spindle.mill.v1.AttemptResult 1336 1336 + 15, // 16: spindle.mill.v1.Message.ack:type_name -> spindle.mill.v1.Ack 1337 1337 + 3, // 17: spindle.mill.v1.NodeSnapshot.EnginesEntry.value:type_name -> spindle.mill.v1.EngineAvailability 1338 1338 + 18, // [18:18] is the sub-list for method output_type 1339 1339 + 18, // [18:18] is the sub-list for method input_type 1340 1340 + 18, // [18:18] is the sub-list for extension type_name 1341 1341 + 18, // [18:18] is the sub-list for extension extendee 1342 1342 + 0, // [0:18] is the sub-list for field type_name 1293 1343 } 1294 1344 1295 1345 func init() { file_spindle_mill_v1_mill_proto_init() } ··· 1302 1352 File: protoimpl.DescBuilder{ 1303 1353 GoPackagePath: reflect.TypeOf(x{}).PkgPath(), 1304 1354 RawDescriptor: unsafe.Slice(unsafe.StringData(file_spindle_mill_v1_mill_proto_rawDesc), len(file_spindle_mill_v1_mill_proto_rawDesc)), 1305 1305 - NumEnums: 0, 1306 1306 - NumMessages: 17, 1355 1355 + NumEnums: 1, 1356 1356 + NumMessages: 18, 1307 1357 NumExtensions: 0, 1308 1358 NumServices: 0, 1309 1359 }, 1310 1360 GoTypes: file_spindle_mill_v1_mill_proto_goTypes, 1311 1361 DependencyIndexes: file_spindle_mill_v1_mill_proto_depIdxs, 1362 1362 + EnumInfos: file_spindle_mill_v1_mill_proto_enumTypes, 1312 1363 MessageInfos: file_spindle_mill_v1_mill_proto_msgTypes, 1313 1364 }.Build() 1314 1365 File_spindle_mill_v1_mill_proto = out.File

+19 -14

spindle/mill/proto/spindle/mill/v1/mill.proto

Reviewed

··· 15 15 string node_id = 2 [(buf.validate.field).string.min_len = 1]; 16 16 // engine names this node can run ("microvm", "nixery"). 17 17 repeated string engines = 3; 18 18 - // GOARCH of the node, so the mill won't place arch-incompatible jobs. 18 18 + // GOARCH of the node, retained as an informational trait for logs. 19 19 string arch = 4; 20 20 // the highest relay offset the executor believes it has sent; a resume hint. 21 21 uint64 last_offset = 5; 22 22 + // opaque operator-defined labels used for runs_on matching. 23 23 + repeated string labels = 6; 22 24 } 23 25 24 26 // Resume is the mill's reply to Hello. The executor replays every buffered ··· 28 30 uint64 ack_offset = 1; 29 31 } 30 32 31 31 - // EngineSnapshot is the changing per-engine state on a node. free_seats is a 32 32 - // coarse "can you take more" hint; 0 also means draining (the mill treats a 33 33 - // draining node as a full one until it leaves). The resource fields are coarse 34 34 - // budget headroom for ranking only; the executor makes the real yes/no call in 35 35 - // ReserveResult. 36 36 - message EngineSnapshot { 37 37 - uint32 free_seats = 1; 38 38 - int64 free_memory_mib = 2; 39 39 - int64 free_vcpus = 3; 40 40 - int64 free_disk_mib = 4; 33 33 + // EngineAvailability is the changing per-engine state on a node. The executor 34 34 + // reports opaque load metrics (higher = more loaded); the mill ranks by the 35 35 + // worst load and uses the sum as a tie-breaker. The mill does not know what 36 36 + // the metric keys mean. 37 37 + message EngineAvailability { 38 38 + bool available = 1; 39 39 + // engine-defined load metrics. keys are opaque to the mill. 40 40 + map<string, double> load = 2; 41 41 } 42 42 43 43 // NodeSnapshot is pushed on connect, periodically, and right after any state ··· 45 45 message NodeSnapshot { 46 46 string node_id = 1; 47 47 uint64 seq = 2; 48 48 - map<string, EngineSnapshot> engines = 3; 48 48 + map<string, EngineAvailability> engines = 3; 49 49 } 50 50 51 51 // ReserveSeat asks an executor to hold a seat for a job. Zero secrets ride this ··· 64 64 uint32 ttl_seconds = 7; 65 65 } 66 66 67 67 + enum RejectClass { 68 68 + REJECT_CLASS_UNSPECIFIED = 0; 69 69 + REJECT_CLASS_TRANSIENT = 1; 70 70 + REJECT_CLASS_INCOMPATIBLE = 2; 71 71 + } 72 72 + 67 73 // ReserveResult is the executor's accept/reject for a ReserveSeat. 68 74 message ReserveResult { 69 75 string lease_id = 1 [(buf.validate.field).string.min_len = 1]; 70 76 bool accepted = 2; 71 77 string reject_reason = 3; 72 72 - // optional bid score the mill ranks accepted leases by (higher is better). 73 73 - double score = 4; 78 78 + RejectClass reject_class = 4; 74 79 } 75 80 76 81 // Secret is a single unlocked secret, sent only inside CommitLease.

+1

spindle/mill/session.go

Reviewed

··· 19 19 // request/response by lease id. We never hold a lock across a decode. 20 20 type millSession struct { 21 21 nodeID string 22 22 + labels []string 22 23 enc messageEncoder 23 24 l *slog.Logger 24 25