+507

spindle/mill/executor/executor.go

Reviewed

··· 1 1 + package executor 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "encoding/json" 6 6 + "fmt" 7 7 + "log/slog" 8 8 + "maps" 9 9 + "net/http" 10 10 + "runtime" 11 11 + "sync" 12 12 + "time" 13 13 + 14 14 + "github.com/bluesky-social/indigo/atproto/syntax" 15 15 + "github.com/gorilla/websocket" 16 16 + 17 17 + "tangled.org/core/api/tangled" 18 18 + "tangled.org/core/notifier" 19 19 + "tangled.org/core/spindle/config" 20 20 + "tangled.org/core/spindle/db" 21 21 + "tangled.org/core/spindle/engine" 22 22 + "tangled.org/core/spindle/models" 23 23 + 24 24 + millproto "tangled.org/core/spindle/mill/proto" 25 25 + millv1 "tangled.org/core/spindle/mill/proto/gen" 26 26 + ) 27 27 + 28 28 + const ( 29 29 + dialBackoffMin = 1 * time.Second 30 30 + dialBackoffMax = 30 * time.Second 31 31 + snapshotEvery = 15 * time.Second 32 32 + defaultSeats = 4 33 33 + ) 34 34 + 35 35 + type Executor struct { 36 36 + millURL string 37 37 + secret string 38 38 + nodeID string 39 39 + seats int 40 40 + 41 41 + engines map[string]models.Engine 42 42 + db *db.DB 43 43 + n *notifier.Notifier 44 44 + cfg *config.Config 45 45 + l *slog.Logger 46 46 + 47 47 + relay *relayLog 48 48 + relayMu sync.Mutex // offset assign + send must be serialized so wire order == offset order 49 49 + 50 50 + connMu sync.Mutex 51 51 + enc messageEncoder 52 52 + 53 53 + mu sync.Mutex 54 54 + active map[string]*reservation 55 55 + draining bool 56 56 + activeCount int 57 57 + } 58 58 + 59 59 + type reservation struct { 60 60 + leaseID string 61 61 + wid models.WorkflowId 62 62 + realEngine models.Engine 63 63 + slot engine.WorkflowSlot 64 64 + wf *models.Workflow 65 65 + repoDid syntax.DID 66 66 + 67 67 + committed bool 68 68 + cancelled bool 69 69 + cancel context.CancelFunc 70 70 + ttlTimer *time.Timer 71 71 + stopTail func() 72 72 + } 73 73 + 74 74 + type messageEncoder interface { 75 75 + Encode(*millproto.Message) error 76 76 + } 77 77 + 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 92 + func New(cfg *config.Config, engines map[string]models.Engine, d *db.DB, n *notifier.Notifier, l *slog.Logger) *Executor { 93 93 + seats := cfg.Mill.Seats 94 94 + if seats <= 0 { 95 95 + seats = defaultSeats 96 96 + } 97 97 + return &Executor{ 98 98 + millURL: cfg.Mill.URL, 99 99 + secret: cfg.Mill.SharedSecret, 100 100 + nodeID: cfg.Server.Hostname, 101 101 + seats: seats, 102 102 + engines: engines, 103 103 + db: d, 104 104 + n: n, 105 105 + cfg: cfg, 106 106 + l: l.With("component", "mill.executor"), 107 107 + relay: newRelayLog(), 108 108 + active: make(map[string]*reservation), 109 109 + } 110 110 + } 111 111 + 112 112 + func (e *Executor) Connect(ctx context.Context) { 113 113 + go e.observeLoop(ctx) 114 114 + 115 115 + backoff := dialBackoffMin 116 116 + for { 117 117 + if ctx.Err() != nil { 118 118 + return 119 119 + } 120 120 + err := e.runSession(ctx) 121 121 + if ctx.Err() != nil { 122 122 + return 123 123 + } 124 124 + e.l.Warn("mill session ended; reconnecting", "err", err, "backoff", backoff) 125 125 + select { 126 126 + case <-ctx.Done(): 127 127 + return 128 128 + case <-time.After(backoff): 129 129 + } 130 130 + backoff = min(backoff*2, dialBackoffMax) 131 131 + } 132 132 + } 133 133 + 134 134 + func (e *Executor) runSession(ctx context.Context) error { 135 135 + header := http.Header{} 136 136 + if e.secret != "" { 137 137 + header.Set("Authorization", "Bearer "+e.secret) 138 138 + } 139 139 + conn, _, err := websocket.DefaultDialer.DialContext(ctx, e.millURL, header) 140 140 + if err != nil { 141 141 + return fmt.Errorf("dial mill: %w", err) 142 142 + } 143 143 + defer conn.Close() 144 144 + 145 145 + stream := millproto.NewWSStream(conn) 146 146 + enc := millproto.NewEncoder(stream) 147 147 + dec := millproto.NewDecoder(stream) 148 148 + 149 149 + hello := &millproto.Message{Hello: &millv1.Hello{ 150 150 + ProtocolVersion: millproto.ProtocolVersion, 151 151 + NodeId: e.nodeID, 152 152 + Engines: e.engineNames(), 153 153 + Arch: runtime.GOARCH, 154 154 + LastOffset: e.relay.lastOffset(), 155 155 + }} 156 156 + if err := enc.Encode(hello); err != nil { 157 157 + return fmt.Errorf("send hello: %w", err) 158 158 + } 159 159 + 160 160 + resumeMsg, err := dec.Decode() 161 161 + if err != nil { 162 162 + return fmt.Errorf("read resume: %w", err) 163 163 + } 164 164 + resume := resumeMsg.GetResume() 165 165 + if resume == nil { 166 166 + return fmt.Errorf("expected resume, got something else") 167 167 + } 168 168 + 169 169 + // install encoder and replay anything the mill missed. 170 170 + e.connMu.Lock() 171 171 + e.enc = enc 172 172 + e.connMu.Unlock() 173 173 + defer func() { 174 174 + e.connMu.Lock() 175 175 + e.enc = nil 176 176 + e.connMu.Unlock() 177 177 + }() 178 178 + 179 179 + e.replay(resume.GetAckOffset()) 180 180 + e.pushSnapshot() 181 181 + e.l.Info("connected to mill", "node", e.nodeID, "resumeFrom", resume.GetAckOffset()) 182 182 + 183 183 + go e.snapshotLoop(ctx, enc) 184 184 + 185 185 + for { 186 186 + msg, err := dec.Decode() 187 187 + if err != nil { 188 188 + return fmt.Errorf("read: %w", err) 189 189 + } 190 190 + e.dispatch(ctx, msg) 191 191 + } 192 192 + } 193 193 + 194 194 + func (e *Executor) replay(ackOffset uint64) { 195 195 + e.relayMu.Lock() 196 196 + defer e.relayMu.Unlock() 197 197 + e.relay.ack(ackOffset) 198 198 + for _, msg := range e.relay.since(ackOffset) { 199 199 + e.send(msg) 200 200 + } 201 201 + } 202 202 + 203 203 + func (e *Executor) send(msg *millproto.Message) { 204 204 + e.connMu.Lock() 205 205 + enc := e.enc 206 206 + e.connMu.Unlock() 207 207 + if enc != nil { 208 208 + _ = enc.Encode(msg) 209 209 + } 210 210 + } 211 211 + 212 212 + func (e *Executor) dispatch(ctx context.Context, msg *millproto.Message) { 213 213 + switch { 214 214 + case msg.GetReserveSeat() != nil: 215 215 + e.handleReserve(ctx, msg.GetReserveSeat()) 216 216 + case msg.GetCommitLease() != nil: 217 217 + e.handleCommit(ctx, msg.GetCommitLease()) 218 218 + case msg.GetReleaseLease() != nil: 219 219 + e.handleRelease(msg.GetReleaseLease().GetLeaseId()) 220 220 + case msg.GetCancelAttempt() != nil: 221 221 + e.handleCancel(msg.GetCancelAttempt().GetLeaseId()) 222 222 + case msg.GetAck() != nil: 223 223 + e.relay.ack(msg.GetAck().GetUpToOffset()) 224 224 + default: 225 225 + e.l.Warn("executor received unexpected message") 226 226 + } 227 227 + } 228 228 + 229 229 + // --- reserve / commit / release / cancel ---------------------------------- 230 230 + 231 231 + func (e *Executor) handleReserve(ctx context.Context, rs *millv1.ReserveSeat) { 232 232 + reject := func(reason string) { 233 233 + e.send(&millproto.Message{ReserveResult: &millv1.ReserveResult{ 234 234 + LeaseId: rs.GetLeaseId(), 235 235 + Accepted: false, 236 236 + RejectReason: reason, 237 237 + }}) 238 238 + } 239 239 + 240 240 + e.mu.Lock() 241 241 + draining := e.draining 242 242 + e.mu.Unlock() 243 243 + if draining { 244 244 + reject("draining") 245 245 + return 246 246 + } 247 247 + 248 248 + realEngine, ok := e.engines[rs.GetTargetEngine()] 249 249 + if !ok { 250 250 + reject("unknown engine " + rs.GetTargetEngine()) 251 251 + return 252 252 + } 253 253 + slotter, ok := realEngine.(engine.WorkflowSlotter) 254 254 + if !ok { 255 255 + reject("engine does not support workflow slots") 256 256 + return 257 257 + } 258 258 + 259 259 + var twf tangled.Pipeline_Workflow 260 260 + if err := json.Unmarshal([]byte(rs.GetRawWorkflowJson()), &twf); err != nil { 261 261 + reject("bad workflow json") 262 262 + return 263 263 + } 264 264 + var tpl tangled.Pipeline 265 265 + if err := json.Unmarshal([]byte(rs.GetRawPipelineJson()), &tpl); err != nil { 266 266 + reject("bad pipeline json") 267 267 + return 268 268 + } 269 269 + 270 270 + pipelineId := models.PipelineId{Knot: rs.GetKnot(), Rkey: rs.GetRkey()} 271 271 + wid := models.WorkflowId{PipelineId: pipelineId, Name: twf.Name} 272 272 + 273 273 + wf, err := realEngine.InitWorkflow(twf, tpl) 274 274 + if err != nil { 275 275 + reject("init workflow: " + err.Error()) 276 276 + return 277 277 + } 278 278 + // the job skipped processPipeline, so inject TANGLED_* env here. 279 279 + if wf.Environment == nil { 280 280 + wf.Environment = make(map[string]string) 281 281 + } 282 282 + maps.Copy(wf.Environment, models.PipelineEnvVars(tpl.TriggerMetadata, pipelineId)) 283 283 + 284 284 + // NoWait: the executor doesn't queue locally; the mill owns the backlog. 285 285 + slot, err := slotter.AcquireWorkflowSlot(ctx, wid, wf, engine.NoWait) 286 286 + if err != nil { 287 287 + reject(err.Error()) 288 288 + return 289 289 + } 290 290 + 291 291 + var repoDid syntax.DID 292 292 + if tpl.TriggerMetadata != nil && tpl.TriggerMetadata.Repo != nil && tpl.TriggerMetadata.Repo.RepoDid != nil { 293 293 + repoDid, _ = syntax.ParseDID(*tpl.TriggerMetadata.Repo.RepoDid) 294 294 + } 295 295 + 296 296 + res := &reservation{ 297 297 + leaseID: rs.GetLeaseId(), 298 298 + wid: wid, 299 299 + realEngine: realEngine, 300 300 + slot: slot, 301 301 + wf: wf, 302 302 + repoDid: repoDid, 303 303 + } 304 304 + res.ttlTimer = time.AfterFunc(ttlDuration(rs.GetTtlSeconds()), func() { e.expireReservation(res.leaseID) }) 305 305 + 306 306 + e.mu.Lock() 307 307 + e.active[res.leaseID] = res 308 308 + e.activeCount++ 309 309 + e.mu.Unlock() 310 310 + 311 311 + e.send(&millproto.Message{ReserveResult: &millv1.ReserveResult{ 312 312 + LeaseId: rs.GetLeaseId(), 313 313 + Accepted: true, 314 314 + Score: e.bidScore(), 315 315 + }}) 316 316 + e.pushSnapshot() 317 317 + } 318 318 + 319 319 + func (e *Executor) handleCommit(ctx context.Context, cl *millv1.CommitLease) { 320 320 + e.mu.Lock() 321 321 + res := e.active[cl.GetLeaseId()] 322 322 + if res == nil { 323 323 + e.mu.Unlock() 324 324 + return 325 325 + } 326 326 + if res.committed { 327 327 + e.mu.Unlock() 328 328 + e.send(&millproto.Message{Committed: &millv1.Committed{LeaseId: cl.GetLeaseId()}}) 329 329 + return 330 330 + } 331 331 + res.committed = true 332 332 + if res.ttlTimer != nil { 333 333 + res.ttlTimer.Stop() 334 334 + } 335 335 + jobCtx, cancel := context.WithCancel(ctx) 336 336 + res.cancel = cancel 337 337 + e.mu.Unlock() 338 338 + 339 339 + vault := newMemVault(cl.GetSecrets()) 340 340 + re := newReservedEngine(res.realEngine, res.slot) 341 341 + pipeline := &models.Pipeline{ 342 342 + RepoDid: res.repoDid, 343 343 + Workflows: map[models.Engine][]models.Workflow{re: {*res.wf}}, 344 344 + } 345 345 + 346 346 + e.startTail(res) 347 347 + 348 348 + go engine.StartWorkflows(e.l, vault, e.cfg, e.db, e.n, jobCtx, pipeline, res.wid.PipelineId) 349 349 + 350 350 + e.send(&millproto.Message{Committed: &millv1.Committed{LeaseId: cl.GetLeaseId()}}) 351 351 + } 352 352 + 353 353 + func (e *Executor) handleRelease(leaseID string) { 354 354 + cleanup, ok := e.takeUncommittedReservation(leaseID, true) 355 355 + if !ok { 356 356 + return 357 357 + } 358 358 + cleanup.run() 359 359 + e.pushSnapshot() 360 360 + } 361 361 + 362 362 + func (e *Executor) handleCancel(leaseID string) { 363 363 + e.mu.Lock() 364 364 + res := e.active[leaseID] 365 365 + if res == nil { 366 366 + e.mu.Unlock() 367 367 + return 368 368 + } 369 369 + res.cancelled = true 370 370 + cancel := res.cancel 371 371 + committed := res.committed 372 372 + var cleanup reservationCleanup 373 373 + if !committed { 374 374 + cleanup = e.removeReservationLocked(res, true) 375 375 + } 376 376 + e.mu.Unlock() 377 377 + 378 378 + if cancel != nil { 379 379 + cancel() 380 380 + } 381 381 + // committed jobs clean up when the observe loop sees the terminal row; an 382 382 + // uncommitted cancel still needs the slot released. 383 383 + if !committed { 384 384 + cleanup.run() 385 385 + e.pushSnapshot() 386 386 + } 387 387 + } 388 388 + 389 389 + func (e *Executor) expireReservation(leaseID string) { 390 390 + cleanup, ok := e.takeUncommittedReservation(leaseID, true) 391 391 + if !ok { 392 392 + return 393 393 + } 394 394 + e.l.Warn("reservation expired before commit", "lease", leaseID) 395 395 + cleanup.run() 396 396 + e.pushSnapshot() 397 397 + } 398 398 + 399 399 + func (e *Executor) takeUncommittedReservation(leaseID string, releaseSlot bool) (reservationCleanup, bool) { 400 400 + e.mu.Lock() 401 401 + defer e.mu.Unlock() 402 402 + res := e.active[leaseID] 403 403 + if res == nil || res.committed { 404 404 + return reservationCleanup{}, false 405 405 + } 406 406 + return e.removeReservationLocked(res, releaseSlot), true 407 407 + } 408 408 + 409 409 + func (e *Executor) finishReservation(res *reservation) (reservationCleanup, bool, bool) { 410 410 + e.mu.Lock() 411 411 + defer e.mu.Unlock() 412 412 + if e.active[res.leaseID] != res { 413 413 + return reservationCleanup{}, false, false 414 414 + } 415 415 + cancelled := res.cancelled 416 416 + return e.removeReservationLocked(res, false), cancelled, true 417 417 + } 418 418 + 419 419 + func (e *Executor) removeReservationLocked(res *reservation, releaseSlot bool) reservationCleanup { 420 420 + delete(e.active, res.leaseID) 421 421 + if e.activeCount > 0 { 422 422 + e.activeCount-- 423 423 + } 424 424 + if res.ttlTimer != nil { 425 425 + res.ttlTimer.Stop() 426 426 + res.ttlTimer = nil 427 427 + } 428 428 + cleanup := reservationCleanup{stopTail: res.stopTail} 429 429 + res.stopTail = nil 430 430 + if releaseSlot { 431 431 + cleanup.slot = res.slot 432 432 + res.slot = nil 433 433 + } 434 434 + return cleanup 435 435 + } 436 436 + 437 437 + // --- snapshots ------------------------------------------------------------- 438 438 + 439 439 + func (e *Executor) snapshotLoop(ctx context.Context, enc *millproto.Encoder) { 440 440 + t := time.NewTicker(snapshotEvery) 441 441 + defer t.Stop() 442 442 + for { 443 443 + select { 444 444 + case <-ctx.Done(): 445 445 + return 446 446 + case <-t.C: 447 447 + // stop pushing once this connection is replaced. 448 448 + e.connMu.Lock() 449 449 + cur := e.enc 450 450 + e.connMu.Unlock() 451 451 + if cur != enc { 452 452 + return 453 453 + } 454 454 + e.pushSnapshot() 455 455 + } 456 456 + } 457 457 + } 458 458 + 459 459 + func (e *Executor) pushSnapshot() { 460 460 + e.mu.Lock() 461 461 + free := uint32(0) 462 462 + if !e.draining { 463 463 + if n := e.seats - e.activeCount; n > 0 { 464 464 + free = uint32(n) 465 465 + } 466 466 + } 467 467 + e.mu.Unlock() 468 468 + 469 469 + engines := make(map[string]*millv1.EngineSnapshot, len(e.engines)) 470 470 + for name := range e.engines { 471 471 + engines[name] = &millv1.EngineSnapshot{FreeSeats: free} 472 472 + } 473 473 + e.send(&millproto.Message{NodeSnapshot: &millv1.NodeSnapshot{ 474 474 + NodeId: e.nodeID, 475 475 + Engines: engines, 476 476 + }}) 477 477 + } 478 478 + 479 479 + func (e *Executor) Drain() { 480 480 + e.mu.Lock() 481 481 + e.draining = true 482 482 + e.mu.Unlock() 483 483 + e.pushSnapshot() 484 484 + } 485 485 + 486 486 + func (e *Executor) bidScore() float64 { 487 487 + e.mu.Lock() 488 488 + defer e.mu.Unlock() 489 489 + return float64(e.seats - e.activeCount) 490 490 + } 491 491 + 492 492 + func (e *Executor) engineNames() []string { 493 493 + names := make([]string, 0, len(e.engines)) 494 494 + for name := range e.engines { 495 495 + names = append(names, name) 496 496 + } 497 497 + return names 498 498 + } 499 499 + 500 500 + func ttlDuration(secs uint32) time.Duration { 501 501 + if secs == 0 { 502 502 + return defaultReservationTTL 503 503 + } 504 504 + return time.Duration(secs) * time.Second 505 505 + } 506 506 + 507 507 + const defaultReservationTTL = 60 * time.Second

+193

spindle/mill/executor/observe.go

Reviewed

··· 1 1 + package executor 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "io" 6 6 + "sync" 7 7 + "time" 8 8 + 9 9 + "github.com/hpcloud/tail" 10 10 + 11 11 + "tangled.org/core/api/tangled" 12 12 + "tangled.org/core/spindle/models" 13 13 + "tangled.org/core/spindle/secrets" 14 14 + 15 15 + millv1 "tangled.org/core/spindle/mill/proto/gen" 16 16 + ) 17 17 + 18 18 + // observeLoop watches the executor's own eventstream and relays the status rows 19 19 + // belonging to active leases. This is the "observe-by-cursor" relay: the jobs 20 20 + // write status to the real db/notifier exactly as a standalone spindle would, 21 21 + // and we forward what is relevant without touching the execution path. 22 22 + func (e *Executor) observeLoop(ctx context.Context) { 23 23 + sub := e.n.Subscribe() 24 24 + defer e.n.Unsubscribe(sub) 25 25 + 26 26 + // start past existing history; only relay rows produced from now on. 27 27 + cursor := time.Now().UnixNano() 28 28 + 29 29 + drain := func() { 30 30 + for { 31 31 + evs, err := e.db.GetEvents(cursor, 100) 32 32 + if err != nil { 33 33 + e.l.Error("observe GetEvents failed", "err", err) 34 34 + return 35 35 + } 36 36 + for _, ev := range evs { 37 37 + cursor = ev.Created 38 38 + if ev.Nsid != tangled.PipelineStatusNSID { 39 39 + continue 40 40 + } 41 41 + st, ok := parseStatus(ev.EventJson) 42 42 + if !ok { 43 43 + continue 44 44 + } 45 45 + e.onStatusRow(st) 46 46 + } 47 47 + if len(evs) < 100 { 48 48 + return 49 49 + } 50 50 + } 51 51 + } 52 52 + 53 53 + // safety tick in case a notify is ever missed. The notifier is the primary 54 54 + // wakeup; this only backstops a dropped signal, so it can be coarse. 55 55 + ticker := time.NewTicker(5 * time.Second) 56 56 + defer ticker.Stop() 57 57 + 58 58 + for { 59 59 + select { 60 60 + case <-ctx.Done(): 61 61 + return 62 62 + case <-sub: 63 63 + drain() 64 64 + case <-ticker.C: 65 65 + drain() 66 66 + } 67 67 + } 68 68 + } 69 69 + 70 70 + func (e *Executor) onStatusRow(st *tangled.PipelineStatus) { 71 71 + res := e.reservationFor(st.Pipeline, st.Workflow) 72 72 + if res == nil { 73 73 + return 74 74 + } 75 75 + 76 76 + kind := models.StatusKind(st.Status) 77 77 + switch { 78 78 + case kind.IsFinish(): 79 79 + e.finishJob(res, st) 80 80 + case kind == models.StatusKindRunning: 81 81 + e.relayMu.Lock() 82 82 + e.send(e.relay.appendStatus(res.leaseID, st, st.Pipeline)) 83 83 + e.relayMu.Unlock() 84 84 + } 85 85 + } 86 86 + 87 87 + // finishJob flushes the log tail, then relays the terminal as an AttemptResult 88 88 + // (which the mill turns into the terminal status row), then forgets the lease. 89 89 + // Tail flush happens first so all log offsets precede the terminal offset. 90 90 + func (e *Executor) finishJob(res *reservation, st *tangled.PipelineStatus) { 91 91 + cleanup, cancelled, ok := e.finishReservation(res) 92 92 + if !ok { 93 93 + return 94 94 + } 95 95 + cleanup.run() 96 96 + 97 97 + e.relayMu.Lock() 98 98 + terminalStatus := st.Status 99 99 + if cancelled { 100 100 + terminalStatus = string(models.StatusKindCancelled) 101 101 + } 102 102 + e.send(e.relay.appendTerminal(res.leaseID, terminalStatus, st)) 103 103 + e.relayMu.Unlock() 104 104 + 105 105 + e.pushSnapshot() 106 106 + } 107 107 + 108 108 + // reservationFor finds the active reservation whose workflow matches a status 109 109 + // row, by pipeline aturi + workflow name. Active reservations are few (bounded 110 110 + // by seats), so a scan is cheaper than maintaining a parallel index. 111 111 + func (e *Executor) reservationFor(pipelineAturi, workflow string) *reservation { 112 112 + e.mu.Lock() 113 113 + defer e.mu.Unlock() 114 114 + for _, res := range e.active { 115 115 + if string(res.wid.PipelineId.AtUri()) == pipelineAturi && res.wid.Name == workflow { 116 116 + return res 117 117 + } 118 118 + } 119 119 + return nil 120 120 + } 121 121 + 122 122 + func (e *Executor) relayLogLine(leaseID, text string) { 123 123 + e.relayMu.Lock() 124 124 + defer e.relayMu.Unlock() 125 125 + e.send(e.relay.appendLog(leaseID, []byte(text))) 126 126 + } 127 127 + 128 128 + // startTail follows the job's local log file and relays each line. The lines are 129 129 + // already-encoded models.LogLine JSON, so we forward them verbatim. 130 130 + func (e *Executor) startTail(res *reservation) { 131 131 + path := models.LogFilePath(e.cfg.Server.LogDir, res.wid) 132 132 + t, err := tail.TailFile(path, tail.Config{ 133 133 + Follow: true, 134 134 + ReOpen: true, 135 135 + MustExist: false, 136 136 + Location: &tail.SeekInfo{Offset: 0, Whence: io.SeekStart}, 137 137 + Logger: tail.DiscardingLogger, 138 138 + }) 139 139 + if err != nil { 140 140 + e.l.Error("tail log file failed", "wid", res.wid, "err", err) 141 141 + return 142 142 + } 143 143 + 144 144 + done := make(chan struct{}) 145 145 + go func() { 146 146 + defer close(done) 147 147 + for line := range t.Lines { 148 148 + if line == nil || line.Err != nil { 149 149 + continue 150 150 + } 151 151 + e.relayLogLine(res.leaseID, line.Text) 152 152 + } 153 153 + }() 154 154 + 155 155 + var once sync.Once 156 156 + res.stopTail = func() { 157 157 + once.Do(func() { 158 158 + // Stop() blocks until tailing ends and t.Lines is closed, so the 159 159 + // consumer goroutine then drains the remaining (finite) buffered 160 160 + // lines and closes done. Waiting on done in full guarantees every 161 161 + // log line is relayed before finishJob sends the terminal — without 162 162 + // it a slow drain lets the terminal overtake a log line, which the 163 163 + // mill then drops against its sealed log file. 164 164 + _ = t.Stop() 165 165 + <-done 166 166 + }) 167 167 + } 168 168 + } 169 169 + 170 170 + // memVault is a temporary in-memory secrets manager holding the secrets the 171 171 + // mill handed over at commit. They never touch disk on the executor. 172 172 + type memVault struct { 173 173 + secrets []secrets.UnlockedSecret 174 174 + } 175 175 + 176 176 + func newMemVault(pb []*millv1.Secret) *memVault { 177 177 + s := make([]secrets.UnlockedSecret, len(pb)) 178 178 + for i, x := range pb { 179 179 + s[i] = secrets.UnlockedSecret{Key: x.GetKey(), Value: x.GetValue()} 180 180 + } 181 181 + return &memVault{secrets: s} 182 182 + } 183 183 + 184 184 + func (v *memVault) GetSecretsUnlocked(ctx context.Context, repo secrets.RepoIdentifier) ([]secrets.UnlockedSecret, error) { 185 185 + return v.secrets, nil 186 186 + } 187 187 + func (v *memVault) GetSecretsLocked(ctx context.Context, repo secrets.RepoIdentifier) ([]secrets.LockedSecret, error) { 188 188 + return nil, nil 189 189 + } 190 190 + func (v *memVault) AddSecret(ctx context.Context, s secrets.UnlockedSecret) error { return nil } 191 191 + func (v *memVault) RemoveSecret(ctx context.Context, s secrets.Secret[any]) error { return nil } 192 192 + 193 193 + var _ secrets.Manager = (*memVault)(nil)

+157

spindle/mill/executor/relay.go

Reviewed

··· 1 1 + package executor 2 2 + 3 3 + import ( 4 4 + "encoding/json" 5 5 + "sync" 6 6 + 7 7 + "tangled.org/core/api/tangled" 8 8 + 9 9 + millproto "tangled.org/core/spindle/mill/proto" 10 10 + millv1 "tangled.org/core/spindle/mill/proto/gen" 11 11 + ) 12 12 + 13 13 + // relayLog is the per-session sequential event log the executor keeps so it can 14 14 + // replay anything the mill missed across a reconnect. Only mill-facing relay 15 15 + // messages (StatusEvent / LogLine / AttemptResult) get offsets and are buffered; 16 16 + // control messages do not. 17 17 + type relayLog struct { 18 18 + mu sync.Mutex 19 19 + entries []*millproto.Message 20 20 + next uint64 // next offset to assign 21 21 + acked uint64 // highest offset the mill has acked 22 22 + } 23 23 + 24 24 + func newRelayLog() *relayLog { 25 25 + return &relayLog{next: 1} 26 26 + } 27 27 + 28 28 + // appendStatus stamps and buffers a relayed running-status event. 29 29 + func (rl *relayLog) appendStatus(leaseID string, st *tangled.PipelineStatus, pipelineAturi string) *millproto.Message { 30 30 + rl.mu.Lock() 31 31 + defer rl.mu.Unlock() 32 32 + offset := rl.next 33 33 + rl.next++ 34 34 + var exit int64 35 35 + if st.ExitCode != nil { 36 36 + exit = *st.ExitCode 37 37 + } 38 38 + var errStr string 39 39 + if st.Error != nil { 40 40 + errStr = *st.Error 41 41 + } 42 42 + msg := &millproto.Message{StatusEvent: &millv1.StatusEvent{ 43 43 + Offset: offset, 44 44 + LeaseId: leaseID, 45 45 + Status: st.Status, 46 46 + Error: errStr, 47 47 + ExitCode: exit, 48 48 + Workflow: st.Workflow, 49 49 + PipelineAturi: pipelineAturi, 50 50 + }} 51 51 + rl.entries = append(rl.entries, msg) 52 52 + return msg 53 53 + } 54 54 + 55 55 + // appendLog stamps and buffers a relayed log line. 56 56 + func (rl *relayLog) appendLog(leaseID string, raw []byte) *millproto.Message { 57 57 + rl.mu.Lock() 58 58 + defer rl.mu.Unlock() 59 59 + offset := rl.next 60 60 + rl.next++ 61 61 + msg := &millproto.Message{LogLine: &millv1.LogLine{ 62 62 + Offset: offset, 63 63 + LeaseId: leaseID, 64 64 + RawJson: raw, 65 65 + }} 66 66 + rl.entries = append(rl.entries, msg) 67 67 + return msg 68 68 + } 69 69 + 70 70 + // appendTerminal stamps and buffers the terminal attempt result. 71 71 + func (rl *relayLog) appendTerminal(leaseID, status string, st *tangled.PipelineStatus) *millproto.Message { 72 72 + rl.mu.Lock() 73 73 + defer rl.mu.Unlock() 74 74 + offset := rl.next 75 75 + rl.next++ 76 76 + var exit int64 77 77 + var errStr string 78 78 + if st != nil { 79 79 + if st.ExitCode != nil { 80 80 + exit = *st.ExitCode 81 81 + } 82 82 + if st.Error != nil { 83 83 + errStr = *st.Error 84 84 + } 85 85 + } 86 86 + msg := &millproto.Message{AttemptResult: &millv1.AttemptResult{ 87 87 + Offset: offset, 88 88 + LeaseId: leaseID, 89 89 + TerminalStatus: status, 90 90 + Error: errStr, 91 91 + ExitCode: exit, 92 92 + }} 93 93 + rl.entries = append(rl.entries, msg) 94 94 + return msg 95 95 + } 96 96 + 97 97 + // ack trims acked entries. 98 98 + func (rl *relayLog) ack(upTo uint64) { 99 99 + rl.mu.Lock() 100 100 + defer rl.mu.Unlock() 101 101 + if upTo > rl.acked { 102 102 + rl.acked = upTo 103 103 + } 104 104 + kept := rl.entries[:0] 105 105 + for _, e := range rl.entries { 106 106 + if offsetOf(e) > rl.acked { 107 107 + kept = append(kept, e) 108 108 + } 109 109 + } 110 110 + // nil the drained tail so acked messages can be GC'd (kept aliases the same 111 111 + // backing array as entries). 112 112 + for i := len(kept); i < len(rl.entries); i++ { 113 113 + rl.entries[i] = nil 114 114 + } 115 115 + rl.entries = kept 116 116 + } 117 117 + 118 118 + // since returns buffered entries with offset strictly greater than offset, for 119 119 + // replay on reconnect. 120 120 + func (rl *relayLog) since(offset uint64) []*millproto.Message { 121 121 + rl.mu.Lock() 122 122 + defer rl.mu.Unlock() 123 123 + var out []*millproto.Message 124 124 + for _, e := range rl.entries { 125 125 + if offsetOf(e) > offset { 126 126 + out = append(out, e) 127 127 + } 128 128 + } 129 129 + return out 130 130 + } 131 131 + 132 132 + func (rl *relayLog) lastOffset() uint64 { 133 133 + rl.mu.Lock() 134 134 + defer rl.mu.Unlock() 135 135 + return rl.next - 1 136 136 + } 137 137 + 138 138 + func offsetOf(m *millproto.Message) uint64 { 139 139 + switch { 140 140 + case m.GetStatusEvent() != nil: 141 141 + return m.GetStatusEvent().GetOffset() 142 142 + case m.GetLogLine() != nil: 143 143 + return m.GetLogLine().GetOffset() 144 144 + case m.GetAttemptResult() != nil: 145 145 + return m.GetAttemptResult().GetOffset() 146 146 + } 147 147 + return 0 148 148 + } 149 149 + 150 150 + // parseStatus decodes a PipelineStatus event row. 151 151 + func parseStatus(raw json.RawMessage) (*tangled.PipelineStatus, bool) { 152 152 + var st tangled.PipelineStatus 153 153 + if err := json.Unmarshal(raw, &st); err != nil { 154 154 + return nil, false 155 155 + } 156 156 + return &st, true 157 157 + }

+66

spindle/mill/executor/reserved.go

Reviewed

··· 1 1 + package executor 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "fmt" 6 6 + "sync" 7 7 + "time" 8 8 + 9 9 + "tangled.org/core/api/tangled" 10 10 + "tangled.org/core/spindle/engine" 11 11 + "tangled.org/core/spindle/models" 12 12 + "tangled.org/core/spindle/secrets" 13 13 + ) 14 14 + 15 15 + // reservedEngine wraps a real engine so that the slot acquired up-front during 16 16 + // ReserveSeat is the slot StartWorkflows gets, instead of acquiring a second 17 17 + // one. Every other method delegates straight to the real engine, so the 18 18 + // execution path runs byte-for-byte as it would standalone. 19 19 + // 20 20 + // This is the only change to the execution path on an executor. 21 21 + type reservedEngine struct { 22 22 + inner models.Engine 23 23 + slot engine.WorkflowSlot 24 24 + once sync.Once 25 25 + } 26 26 + 27 27 + // newReservedEngine returns a wrapper around inner that hands back slot exactly 28 28 + // once from AcquireWorkflowSlot. 29 29 + 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) 51 51 + } 52 52 + 53 53 + // AcquireWorkflowSlot hands back the pre-acquired slot exactly once. The slot 54 54 + // was already obtained (and resource-accounted) during ReserveSeat, so a second 55 55 + // acquire would double-count. 56 56 + func (e *reservedEngine) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, _ engine.AcquireMode) (engine.WorkflowSlot, error) { 57 57 + var slot engine.WorkflowSlot 58 58 + e.once.Do(func() { 59 59 + slot = e.slot 60 60 + e.slot = nil 61 61 + }) 62 62 + if slot == nil { 63 63 + return nil, fmt.Errorf("reserved slot already consumed") 64 64 + } 65 65 + return slot, nil 66 66 + }

+148

spindle/mill/executor/reserved_test.go

Reviewed

··· 1 1 + package executor 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "io" 6 6 + "log/slog" 7 7 + "testing" 8 8 + "time" 9 9 + 10 10 + "tangled.org/core/api/tangled" 11 11 + "tangled.org/core/spindle/engine" 12 12 + "tangled.org/core/spindle/mill/proto" 13 13 + millv1 "tangled.org/core/spindle/mill/proto/gen" 14 14 + "tangled.org/core/spindle/models" 15 15 + "tangled.org/core/spindle/secrets" 16 16 + ) 17 17 + 18 18 + type captureEncoder struct { 19 19 + messages chan *millproto.Message 20 20 + } 21 21 + 22 22 + func newCaptureEncoder() *captureEncoder { 23 23 + return &captureEncoder{messages: make(chan *millproto.Message, 4)} 24 24 + } 25 25 + 26 26 + func (e *captureEncoder) Encode(msg *millproto.Message) error { 27 27 + e.messages <- msg 28 28 + return nil 29 29 + } 30 30 + 31 31 + type fakeSlot struct{ released int } 32 32 + 33 33 + func (s *fakeSlot) Release() { s.released++ } 34 34 + 35 35 + // fakeEngine records which methods got called and never acquires a slot of its 36 36 + // own; the wrapper must supply the slot. 37 37 + type fakeEngine struct { 38 38 + setupCalled bool 39 39 + runCalled bool 40 40 + destroyCalled bool 41 41 + acquireCalled bool 42 42 + } 43 43 + 44 44 + func (e *fakeEngine) InitWorkflow(twf tangled.Pipeline_Workflow, tpl tangled.Pipeline) (*models.Workflow, error) { 45 45 + return &models.Workflow{Name: twf.Name}, nil 46 46 + } 47 47 + func (e *fakeEngine) SetupWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, l models.WorkflowLogger) error { 48 48 + e.setupCalled = true 49 49 + return nil 50 50 + } 51 51 + func (e *fakeEngine) WorkflowTimeout() time.Duration { return 7 * time.Minute } 52 52 + func (e *fakeEngine) DestroyWorkflow(ctx context.Context, wid models.WorkflowId) error { 53 53 + e.destroyCalled = true 54 54 + return nil 55 55 + } 56 56 + func (e *fakeEngine) RunStep(ctx context.Context, wid models.WorkflowId, w *models.Workflow, idx int, s []secrets.UnlockedSecret, l models.WorkflowLogger) error { 57 57 + e.runCalled = true 58 58 + return nil 59 59 + } 60 60 + 61 61 + func (e *fakeEngine) AcquireWorkflowSlot(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, mode engine.AcquireMode) (engine.WorkflowSlot, error) { 62 62 + e.acquireCalled = true 63 63 + return engine.NoopSlot{}, nil 64 64 + } 65 65 + 66 66 + func TestReservedEngineHandsBackHeldSlotOnce(t *testing.T) { 67 67 + inner := &fakeEngine{} 68 68 + slot := &fakeSlot{} 69 69 + re := newReservedEngine(inner, slot) 70 70 + 71 71 + got, err := re.(engine.WorkflowSlotter).AcquireWorkflowSlot(context.Background(), models.WorkflowId{}, nil, engine.Wait) 72 72 + if err != nil { 73 73 + t.Fatalf("AcquireWorkflowSlot() error = %v", err) 74 74 + } 75 75 + if got != engine.WorkflowSlot(slot) { 76 76 + t.Fatal("AcquireWorkflowSlot() did not return the held slot") 77 77 + } 78 78 + if inner.acquireCalled { 79 79 + t.Fatal("wrapper must not call the inner engine's AcquireWorkflowSlot") 80 80 + } 81 81 + 82 82 + if _, err := re.(engine.WorkflowSlotter).AcquireWorkflowSlot(context.Background(), models.WorkflowId{}, nil, engine.Wait); err == nil { 83 83 + t.Fatal("second AcquireWorkflowSlot() should error") 84 84 + } 85 85 + } 86 86 + 87 87 + func TestReservedEngineForwardsLifecycle(t *testing.T) { 88 88 + inner := &fakeEngine{} 89 89 + re := newReservedEngine(inner, &fakeSlot{}) 90 90 + 91 91 + if d := re.WorkflowTimeout(); d != 7*time.Minute { 92 92 + t.Fatalf("WorkflowTimeout() = %v, want 7m", d) 93 93 + } 94 94 + if err := re.SetupWorkflow(context.Background(), models.WorkflowId{}, &models.Workflow{}, models.NullLogger{}); err != nil { 95 95 + t.Fatal(err) 96 96 + } 97 97 + if err := re.RunStep(context.Background(), models.WorkflowId{}, &models.Workflow{Steps: []models.Step{}}, 0, nil, models.NullLogger{}); err != nil { 98 98 + t.Fatal(err) 99 99 + } 100 100 + if err := re.DestroyWorkflow(context.Background(), models.WorkflowId{}); err != nil { 101 101 + t.Fatal(err) 102 102 + } 103 103 + if !inner.setupCalled || !inner.runCalled || !inner.destroyCalled { 104 104 + t.Fatalf("lifecycle not forwarded: %+v", inner) 105 105 + } 106 106 + } 107 107 + 108 108 + func TestHandleCommitIsIdempotent(t *testing.T) { 109 109 + enc := newCaptureEncoder() 110 110 + 111 111 + e := &Executor{ 112 112 + l: slog.New(slog.NewTextHandler(io.Discard, nil)), 113 113 + enc: enc, 114 114 + active: map[string]*reservation{"lease-1": {leaseID: "lease-1", committed: true}}, 115 115 + } 116 116 + 117 117 + e.handleCommit(context.Background(), &millv1.CommitLease{LeaseId: "lease-1"}) 118 118 + msg := <-enc.messages 119 119 + if got := msg.GetCommitted().GetLeaseId(); got != "lease-1" { 120 120 + t.Fatalf("Committed lease = %q, want lease-1", got) 121 121 + } 122 122 + } 123 123 + 124 124 + func TestFinishJobReportsCancelledReservationAsCancelled(t *testing.T) { 125 125 + res := &reservation{ 126 126 + leaseID: "lease-1", 127 127 + wid: models.WorkflowId{PipelineId: models.PipelineId{Knot: "k", Rkey: "r"}, Name: "build"}, 128 128 + cancelled: true, 129 129 + } 130 130 + e := &Executor{ 131 131 + relay: newRelayLog(), 132 132 + active: map[string]*reservation{res.leaseID: res}, 133 133 + } 134 134 + 135 135 + e.finishJob(res, &tangled.PipelineStatus{ 136 136 + Pipeline: string(res.wid.PipelineId.AtUri()), 137 137 + Workflow: res.wid.Name, 138 138 + Status: string(models.StatusKindFailed), 139 139 + }) 140 140 + 141 141 + if len(e.relay.entries) != 1 { 142 142 + t.Fatalf("relay entries = %d, want 1", len(e.relay.entries)) 143 143 + } 144 144 + got := e.relay.entries[0].GetAttemptResult().GetTerminalStatus() 145 145 + if got != string(models.StatusKindCancelled) { 146 146 + t.Fatalf("terminal status = %q, want cancelled", got) 147 147 + } 148 148 + }