runner.go

v0.6.1

package legatus

import (
	"context"
	"io"
	"log/slog"
	"sync"
	"time"

	"go.bigb.es/auxilia/culpa"
	"go.bigb.es/auxilia/scribe"
)

// Runner executes flows across an inventory. It pools one Conn per host and
// carries a SharedStore so sequential Run calls can pass data between groups
// (e.g. exits → relay). The Runner reaches hosts only through the Dialer/Conn
// interfaces, never SSH directly (IV1, PC1).
type Runner struct {
	inv         *Inventory
	dialer      Dialer
	concurrency int
	policy      ErrorPolicy
	reporter    Reporter
	prompter    Prompter
	logger      *slog.Logger
	dryRun      bool
	out         io.Writer

	shared *SharedStore

	mu   sync.Mutex
	pool map[string]Conn
}

// NewRunner builds a Runner. Defaults: concurrency 1, ContinueOnError, no-op
// reporter, non-interactive prompter, slog.Default logger.
func NewRunner(inv *Inventory, opts ...Option) *Runner {
	r := &Runner{
		inv:         inv,
		concurrency: 1,
		policy:      ContinueOnError,
		reporter:    NopReporter{},
		prompter:    DenyPrompter{},
		logger:      slog.Default(),
		out:         io.Discard,
		shared:      NewSharedStore(),
		pool:        map[string]Conn{},
	}
	for _, o := range opts {
		o(r)
	}
	return r
}

// Shared returns the cross-host store, persistent across Run calls.
func (r *Runner) Shared() *SharedStore { return r.shared }

// Run executes flow against the hosts matching sel (all hosts if none), fanning
// out across an ordered worker pool capped at the configured concurrency. It
// always returns a result per host. Under FailFast, once a host fails the run
// context is cancelled and hosts not yet dispatched are recorded as cancelled
// without dialing — because jobs are fed in inventory order, with concurrency 1
// this cancellation is deterministic.
func (r *Runner) Run(ctx context.Context, src FlowSource, sel ...Selector) RunResult {
	flow, err := src.Resolve()
	if err != nil {
		rr := r.resolveFailure(flow.Name, err)
		r.reporter.RunDone(rr)
		return rr
	}

	hosts := r.inv.Hosts(sel...)
	results := make([]FlowResult, len(hosts))

	conc := r.concurrency
	if conc < 1 {
		conc = 1
	}
	if conc > len(hosts) {
		conc = len(hosts)
	}

	runCtx, cancel := context.WithCancel(ctx)
	defer cancel()

	type job struct {
		idx int
		h   *Host
	}
	jobs := make(chan job)
	var wg sync.WaitGroup
	var failOnce sync.Once

	for w := 0; w < conc; w++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			for j := range jobs {
				fr := r.runHost(runCtx, flow, j.h)
				results[j.idx] = fr
				if r.policy == FailFast && fr.Status == StatusFailed {
					failOnce.Do(cancel)
				}
			}
		}()
	}

	for i, h := range hosts {
		if r.policy == FailFast {
			select {
			case <-runCtx.Done():
				results[i] = r.cancelledFlow(flow, h, runCtx.Err())
				continue
			default:
			}
		}
		jobs <- job{idx: i, h: h}
	}
	close(jobs)
	wg.Wait()

	rr := RunResult{Flows: results}
	r.reporter.RunDone(rr)
	return rr
}

// resolveFailure builds the RunResult for a FlowSource that failed to resolve
// before any host fan-out: a single host-less FlowResult carrying the error under
// a synthetic "resolve" step, so RunResult.Failed() is true and the error reaches
// the caller through the same res.Failed() path as a step failure.
func (r *Runner) resolveFailure(name string, err error) RunResult {
	sr := StepResult{Step: "resolve", Status: StatusFailed, Err: err, Started: time.Now()}
	fr := FlowResult{Flow: name, Steps: []StepResult{sr}, Status: StatusFailed, Params: map[string]any{}}
	return RunResult{Flows: []FlowResult{fr}}
}

func (r *Runner) cancelledFlow(flow Flow, h *Host, err error) FlowResult {
	sr := StepResult{Step: "cancelled", Status: StatusFailed, Err: err, Started: time.Now()}
	return FlowResult{Flow: flow.Name, Host: h.Name, Steps: []StepResult{sr}, Status: StatusFailed, Params: map[string]any{}}
}

func (r *Runner) runHost(ctx context.Context, flow Flow, h *Host) FlowResult {
	hlog := r.logger.With("host", h.Name, "flow", flow.Name)
	r.reporter.FlowStart(h.Name, flow.Name)
	fr := FlowResult{Flow: flow.Name, Host: h.Name}

	conn, err := r.connFor(ctx, h)
	if err != nil {
		err = culpa.Wrap(err, "dial host "+h.Name)
		sr := StepResult{Step: "dial", Status: StatusFailed, Err: err, Started: time.Now()}
		hlog.Error("dial failed", scribe.Err(err))
		r.reporter.StepDone(h.Name, sr)
		fr.Steps = append(fr.Steps, sr)
		fr.Status = StatusFailed
		fr.Params = map[string]any{}
		r.reporter.FlowDone(h.Name, fr)
		return fr
	}

	fc := NewFlowCtx(ctx, h, conn,
		WithFlowLogger(hlog),
		WithFlowDryRun(r.dryRun),
		WithFlowPrompter(r.prompter),
		WithFlowShared(r.shared),
		WithFlowOutput(r.out),
	)

	for _, step := range flow.Steps {
		sr := r.runStep(fc, hlog, h.Name, step)
		// Record change before the next step's Check so later steps can gate on it
		// (steps.When/OnChanged); WouldChange counts so gates fire in dry-run too.
		fc.setChanged(sr.Step, sr.Status == StatusChanged || sr.Status == StatusWouldChange)
		fr.Steps = append(fr.Steps, sr)
		if sr.Status == StatusFailed {
			break // fail-fast within a flow (IV7)
		}
	}
	fr.Status = aggregate(fr.Steps)
	fr.Params = fc.snapshot()
	r.reporter.FlowDone(h.Name, fr)
	return fr
}

// runStep runs one step's check/apply lifecycle, emitting structured logs and
// reporter events with no step-author involvement (PC5).
func (r *Runner) runStep(fc *FlowCtx, hlog *slog.Logger, host string, step Step) StepResult {
	name := step.Name()
	slogger := hlog.With("step", name)
	fc.log = slogger // per-step tag; params and other fields persist across steps
	r.reporter.StepStart(host, name)

	started := time.Now()
	slogger.Debug("step starting")
	sr := StepResult{Step: name, Started: started}

	need, err := step.Check(fc)
	if err != nil {
		sr.Status = StatusFailed
		sr.Err = culpa.Wrap(err, "check "+name)
		sr.Duration = time.Since(started)
		slogger.Error("step failed", "stage", "check", scribe.Err(sr.Err))
		r.reporter.StepDone(host, sr)
		return sr
	}
	slogger.Debug("check complete", "need", need)

	if !need {
		sr.Status = StatusSkipped
		sr.Duration = time.Since(started)
		slogger.Info("step skipped")
		r.reporter.StepDone(host, sr)
		return sr
	}
	if fc.dryRun {
		sr.Status = StatusWouldChange
		sr.Duration = time.Since(started)
		slogger.Info("step would change (dry-run)")
		r.reporter.StepDone(host, sr)
		return sr
	}

	out, err := step.Apply(fc)
	sr.Duration = time.Since(started)
	if err != nil {
		sr.Status = StatusFailed
		sr.Err = culpa.Wrap(err, "apply "+name)
		slogger.Error("step failed", "stage", "apply", scribe.Err(sr.Err))
		r.reporter.StepDone(host, sr)
		return sr
	}
	sr.Outcome = out
	if out.Changed {
		sr.Status = StatusChanged
	} else {
		sr.Status = StatusOK
	}
	slogger.Info("step applied", "status", sr.Status.String(), "changed", out.Changed, "dur", sr.Duration)
	r.reporter.StepDone(host, sr)
	return sr
}

func (r *Runner) connFor(ctx context.Context, h *Host) (Conn, error) {
	r.mu.Lock()
	c, ok := r.pool[h.Name]
	r.mu.Unlock()
	if ok {
		return c, nil
	}
	if r.dialer == nil {
		return nil, culpa.WithCode(culpa.New("legatus: no dialer configured"), "LEGATUS_NO_DIALER")
	}
	c, err := r.dialer.Dial(ctx, h)
	if err != nil {
		return nil, err
	}
	r.mu.Lock()
	if existing, ok := r.pool[h.Name]; ok { // lost a race; keep the first
		r.mu.Unlock()
		_ = c.Close()
		return existing, nil
	}
	r.pool[h.Name] = c
	r.mu.Unlock()
	return c, nil
}

// Close closes all pooled connections. It also satisfies steward's Stopper.
func (r *Runner) Close() error {
	r.mu.Lock()
	defer r.mu.Unlock()
	var firstErr error
	for name, c := range r.pool {
		if err := c.Close(); err != nil && firstErr == nil {
			firstErr = culpa.Wrap(err, "close conn "+name)
		}
		delete(r.pool, name)
	}
	return firstErr
}

// Stop closes pooled connections (steward Stopper).
func (r *Runner) Stop(context.Context) error { return r.Close() }


1	package legatus
2
3	import (
4	"context"
5	"io"
6	"log/slog"
7	"sync"
8	"time"
9
10	"go.bigb.es/auxilia/culpa"
11	"go.bigb.es/auxilia/scribe"
12	)
13
14	// Runner executes flows across an inventory. It pools one Conn per host and
15	// carries a SharedStore so sequential Run calls can pass data between groups
16	// (e.g. exits → relay). The Runner reaches hosts only through the Dialer/Conn
17	// interfaces, never SSH directly (IV1, PC1).
18	type Runner struct {
19	inv *Inventory
20	dialer Dialer
21	concurrency int
22	policy ErrorPolicy
23	reporter Reporter
24	prompter Prompter
25	logger *slog.Logger
26	dryRun bool
27	out io.Writer
28
29	shared *SharedStore
30
31	mu sync.Mutex
32	pool map[string]Conn
33	}
34
35	// NewRunner builds a Runner. Defaults: concurrency 1, ContinueOnError, no-op
36	// reporter, non-interactive prompter, slog.Default logger.
37	func NewRunner(inv Inventory, opts ...Option) Runner {
38	r := &Runner{
39	inv: inv,
40	concurrency: 1,
41	policy: ContinueOnError,
42	reporter: NopReporter{},
43	prompter: DenyPrompter{},
44	logger: slog.Default(),
45	out: io.Discard,
46	shared: NewSharedStore(),
47	pool: map[string]Conn{},
48	}
49	for _, o := range opts {
50	o(r)
51	}
52	return r
53	}
54
55	// Shared returns the cross-host store, persistent across Run calls.
56	func (r Runner) Shared() SharedStore { return r.shared }
57
58	// Run executes flow against the hosts matching sel (all hosts if none), fanning
59	// out across an ordered worker pool capped at the configured concurrency. It
60	// always returns a result per host. Under FailFast, once a host fails the run
61	// context is cancelled and hosts not yet dispatched are recorded as cancelled
62	// without dialing — because jobs are fed in inventory order, with concurrency 1
63	// this cancellation is deterministic.
64	func (r *Runner) Run(ctx context.Context, src FlowSource, sel ...Selector) RunResult {
65	flow, err := src.Resolve()
66	if err != nil {
67	rr := r.resolveFailure(flow.Name, err)
68	r.reporter.RunDone(rr)
69	return rr
70	}
71
72	hosts := r.inv.Hosts(sel...)
73	results := make([]FlowResult, len(hosts))
74
75	conc := r.concurrency
76	if conc < 1 {
77	conc = 1
78	}
79	if conc > len(hosts) {
80	conc = len(hosts)
81	}
82
83	runCtx, cancel := context.WithCancel(ctx)
84	defer cancel()
85
86	type job struct {
87	idx int
88	h *Host
89	}
90	jobs := make(chan job)
91	var wg sync.WaitGroup
92	var failOnce sync.Once
93
94	for w := 0; w < conc; w++ {
95	wg.Add(1)
96	go func() {
97	defer wg.Done()
98	for j := range jobs {
99	fr := r.runHost(runCtx, flow, j.h)
100	results[j.idx] = fr
101	if r.policy == FailFast && fr.Status == StatusFailed {
102	failOnce.Do(cancel)
103	}
104	}
105	}()
106	}
107
108	for i, h := range hosts {
109	if r.policy == FailFast {
110	select {
111	case <-runCtx.Done():
112	results[i] = r.cancelledFlow(flow, h, runCtx.Err())
113	continue
114	default:
115	}
116	}
117	jobs <- job{idx: i, h: h}
118	}
119	close(jobs)
120	wg.Wait()
121
122	rr := RunResult{Flows: results}
123	r.reporter.RunDone(rr)
124	return rr
125	}
126
127	// resolveFailure builds the RunResult for a FlowSource that failed to resolve
128	// before any host fan-out: a single host-less FlowResult carrying the error under
129	// a synthetic "resolve" step, so RunResult.Failed() is true and the error reaches
130	// the caller through the same res.Failed() path as a step failure.
131	func (r *Runner) resolveFailure(name string, err error) RunResult {
132	sr := StepResult{Step: "resolve", Status: StatusFailed, Err: err, Started: time.Now()}
133	fr := FlowResult{Flow: name, Steps: []StepResult{sr}, Status: StatusFailed, Params: map[string]any{}}
134	return RunResult{Flows: []FlowResult{fr}}
135	}
136
137	func (r Runner) cancelledFlow(flow Flow, h Host, err error) FlowResult {
138	sr := StepResult{Step: "cancelled", Status: StatusFailed, Err: err, Started: time.Now()}
139	return FlowResult{Flow: flow.Name, Host: h.Name, Steps: []StepResult{sr}, Status: StatusFailed, Params: map[string]any{}}
140	}
141
142	func (r Runner) runHost(ctx context.Context, flow Flow, h Host) FlowResult {
143	hlog := r.logger.With("host", h.Name, "flow", flow.Name)
144	r.reporter.FlowStart(h.Name, flow.Name)
145	fr := FlowResult{Flow: flow.Name, Host: h.Name}
146
147	conn, err := r.connFor(ctx, h)
148	if err != nil {
149	err = culpa.Wrap(err, "dial host "+h.Name)
150	sr := StepResult{Step: "dial", Status: StatusFailed, Err: err, Started: time.Now()}
151	hlog.Error("dial failed", scribe.Err(err))
152	r.reporter.StepDone(h.Name, sr)
153	fr.Steps = append(fr.Steps, sr)
154	fr.Status = StatusFailed
155	fr.Params = map[string]any{}
156	r.reporter.FlowDone(h.Name, fr)
157	return fr
158	}
159
160	fc := NewFlowCtx(ctx, h, conn,
161	WithFlowLogger(hlog),
162	WithFlowDryRun(r.dryRun),
163	WithFlowPrompter(r.prompter),
164	WithFlowShared(r.shared),
165	WithFlowOutput(r.out),
166	)
167
168	for _, step := range flow.Steps {
169	sr := r.runStep(fc, hlog, h.Name, step)
170	// Record change before the next step's Check so later steps can gate on it
171	// (steps.When/OnChanged); WouldChange counts so gates fire in dry-run too.
172	fc.setChanged(sr.Step, sr.Status == StatusChanged \|\| sr.Status == StatusWouldChange)
173	fr.Steps = append(fr.Steps, sr)
174	if sr.Status == StatusFailed {
175	break // fail-fast within a flow (IV7)
176	}
177	}
178	fr.Status = aggregate(fr.Steps)
179	fr.Params = fc.snapshot()
180	r.reporter.FlowDone(h.Name, fr)
181	return fr
182	}
183
184	// runStep runs one step's check/apply lifecycle, emitting structured logs and
185	// reporter events with no step-author involvement (PC5).
186	func (r Runner) runStep(fc FlowCtx, hlog *slog.Logger, host string, step Step) StepResult {
187	name := step.Name()
188	slogger := hlog.With("step", name)
189	fc.log = slogger // per-step tag; params and other fields persist across steps
190	r.reporter.StepStart(host, name)
191
192	started := time.Now()
193	slogger.Debug("step starting")
194	sr := StepResult{Step: name, Started: started}
195
196	need, err := step.Check(fc)
197	if err != nil {
198	sr.Status = StatusFailed
199	sr.Err = culpa.Wrap(err, "check "+name)
200	sr.Duration = time.Since(started)
201	slogger.Error("step failed", "stage", "check", scribe.Err(sr.Err))
202	r.reporter.StepDone(host, sr)
203	return sr
204	}
205	slogger.Debug("check complete", "need", need)
206
207	if !need {
208	sr.Status = StatusSkipped
209	sr.Duration = time.Since(started)
210	slogger.Info("step skipped")
211	r.reporter.StepDone(host, sr)
212	return sr
213	}
214	if fc.dryRun {
215	sr.Status = StatusWouldChange
216	sr.Duration = time.Since(started)
217	slogger.Info("step would change (dry-run)")
218	r.reporter.StepDone(host, sr)
219	return sr
220	}
221
222	out, err := step.Apply(fc)
223	sr.Duration = time.Since(started)
224	if err != nil {
225	sr.Status = StatusFailed
226	sr.Err = culpa.Wrap(err, "apply "+name)
227	slogger.Error("step failed", "stage", "apply", scribe.Err(sr.Err))
228	r.reporter.StepDone(host, sr)
229	return sr
230	}
231	sr.Outcome = out
232	if out.Changed {
233	sr.Status = StatusChanged
234	} else {
235	sr.Status = StatusOK
236	}
237	slogger.Info("step applied", "status", sr.Status.String(), "changed", out.Changed, "dur", sr.Duration)
238	r.reporter.StepDone(host, sr)
239	return sr
240	}
241
242	func (r Runner) connFor(ctx context.Context, h Host) (Conn, error) {
243	r.mu.Lock()
244	c, ok := r.pool[h.Name]
245	r.mu.Unlock()
246	if ok {
247	return c, nil
248	}
249	if r.dialer == nil {
250	return nil, culpa.WithCode(culpa.New("legatus: no dialer configured"), "LEGATUS_NO_DIALER")
251	}
252	c, err := r.dialer.Dial(ctx, h)
253	if err != nil {
254	return nil, err
255	}
256	r.mu.Lock()
257	if existing, ok := r.pool[h.Name]; ok { // lost a race; keep the first
258	r.mu.Unlock()
259	_ = c.Close()
260	return existing, nil
261	}
262	r.pool[h.Name] = c
263	r.mu.Unlock()
264	return c, nil
265	}
266
267	// Close closes all pooled connections. It also satisfies steward's Stopper.
268	func (r *Runner) Close() error {
269	r.mu.Lock()
270	defer r.mu.Unlock()
271	var firstErr error
272	for name, c := range r.pool {
273	if err := c.Close(); err != nil && firstErr == nil {
274	firstErr = culpa.Wrap(err, "close conn "+name)
275	}
276	delete(r.pool, name)
277	}
278	return firstErr
279	}
280
281	// Stop closes pooled connections (steward Stopper).
282	func (r *Runner) Stop(context.Context) error { return r.Close() }
283

runner.go

Source Files