60 lines
1.5 KiB
Go
60 lines
1.5 KiB
Go
package task
|
|
|
|
import (
|
|
"fmt"
|
|
|
|
"git.yeij.top/AskaEth/Cyrene_v2/internal/runtime/exec"
|
|
)
|
|
|
|
// run executes one task on its own goroutine. It always returns through finish,
|
|
// so the done channel is closed and the goroutine never leaks even on a panic.
|
|
func (m *Manager) run(t *Task, req exec.ExecRequest) {
|
|
m.active.Add(1)
|
|
defer m.active.Add(-1)
|
|
defer func() {
|
|
if r := recover(); r != nil {
|
|
m.log.Error("task run panicked", "id", t.ID, "panic", r)
|
|
t.finish(nil, fmt.Errorf("task: run panicked: %v", r))
|
|
}
|
|
}()
|
|
|
|
res, err := m.prov.Execute(t.ctx, req)
|
|
t.finish(res, err)
|
|
}
|
|
|
|
// publishOutput fans the current output out on "task.output".
|
|
func (m *Manager) publishOutput(t *Task) {
|
|
s := t.Snapshot()
|
|
m.publish("task.output", OutputEvent{
|
|
ID: s.ID,
|
|
Kind: s.Kind,
|
|
Status: s.Status,
|
|
Output: s.Output,
|
|
Truncated: s.Truncated,
|
|
})
|
|
}
|
|
|
|
// publishDone fans a task's terminal state out on "task.done".
|
|
func (m *Manager) publishDone(t *Task) {
|
|
s := t.Snapshot()
|
|
m.publish("task.done", DoneEvent{
|
|
ID: s.ID,
|
|
Kind: s.Kind,
|
|
Status: s.Status,
|
|
Output: s.Output,
|
|
ExitCode: s.ExitCode,
|
|
Truncated: s.Truncated,
|
|
Error: s.Err,
|
|
StartedAt: s.StartedAt,
|
|
FinishedAt: s.FinishedAt,
|
|
})
|
|
}
|
|
|
|
// publish relays an event to the manager bus, logging a warning (never panicking)
|
|
// when the bus rejects it, e.g. after Close.
|
|
func (m *Manager) publish(subject string, payload any) {
|
|
if err := m.bus.Publish(m.ctx, subject, payload); err != nil {
|
|
m.log.Warn("task bus publish failed", "subject", subject, "err", err)
|
|
}
|
|
}
|