1// SPDX-License-Identifier: MIT
2
3package host
4
5import (
6 "context"
7 "sync"
8 "time"
9
10 "connectrpc.com/connect"
11 "github.com/bnema/zerowrap"
12 "google.golang.org/protobuf/types/known/timestamppb"
13
14 ermite "bnema.dev/ermite-api/gen/ermite"
15)
16
17// Limits of AppendLog (runner.proto): at most 1,000 lines and 64 KiB of text
18// per request, all in the same step.
19const (
20 maxBatchLines = 1000
21 maxBatchBytes = 64 * 1024
22 // maxQueuedBytes is the log kept in memory while the server cannot be
23 // reached; beyond it the executor is slowed down (the stream is not read)
24 // rather than the host's memory filled.
25 maxQueuedBytes = 16 << 20
26)
27
28type logLine struct {
29 step int32
30 unixNano int64
31 text string
32 continued bool
33}
34
35// event is something to send, in the order it happened: a log line, or a
36// snapshot of a step (UpdateJob).
37type event struct {
38 line *logLine
39 step *ermite.StepUpdate
40}
41
42// outbox sends logs and step updates of one job to ermite, in order, one call
43// at a time. Log lines are numbered from 0 across the whole job; the number of
44// the next line is the offset of the next AppendLog, so that a retry after a
45// lost response sends the same lines at the same offset (the server skips what
46// it already holds).
47type outbox struct {
48 client callClient
49 jobID string
50 t Timing
51 // onEnded is called once when the server says the job is over (JobEnded)
52 // or the assignment is gone; the outbox then drops everything.
53 onEnded func(error)
54 log zerowrap.Logger
55
56 mu sync.Mutex
57 cond *sync.Cond
58 events []event
59 queued int // bytes of log text in events
60 closing bool
61 dead bool
62 done chan struct{}
63
64 offset int64 // lines the server holds, as far as we know
65 // offsetMaybeStale is set when a send was cut short by its context while
66 // the call may have been applied by the server with the response lost: the
67 // offset is then possibly behind the server's count.
68 offsetMaybeStale bool
69}
70
71// callClient is the part of the RunnerService client the job uses.
72type callClient interface {
73 UpdateJob(context.Context, *connect.Request[ermite.UpdateJobRequest]) (*connect.Response[ermite.UpdateJobResponse], error)
74 AppendLog(context.Context, *connect.Request[ermite.AppendLogRequest]) (*connect.Response[ermite.AppendLogResponse], error)
75}
76
77func newOutbox(c callClient, jobID string, t Timing, onEnded func(error), log zerowrap.Logger) *outbox {
78 o := &outbox{client: c, jobID: jobID, t: t, onEnded: onEnded, log: log, done: make(chan struct{})}
79 o.cond = sync.NewCond(&o.mu)
80 return o
81}
82
83// start runs the sender until Close or until ctx ends.
84func (o *outbox) start(ctx context.Context) {
85 go func() {
86 defer close(o.done)
87 o.loop(ctx)
88 }()
89 // A cancelled context must wake the loop and the producers.
90 go func() {
91 select {
92 case <-ctx.Done():
93 o.kill()
94 case <-o.done:
95 }
96 }()
97}
98
99func (o *outbox) kill() {
100 o.mu.Lock()
101 o.dead = true
102 o.events = nil
103 o.queued = 0
104 o.mu.Unlock()
105 o.cond.Broadcast()
106}
107
108// Log queues a line of the job log. It blocks while too much is queued.
109func (o *outbox) Log(l logLine) {
110 o.mu.Lock()
111 for o.queued > maxQueuedBytes && !o.dead && !o.closing {
112 o.cond.Wait()
113 }
114 if o.dead {
115 o.mu.Unlock()
116 return
117 }
118 o.events = append(o.events, event{line: &l})
119 o.queued += len(l.text) + 32
120 o.mu.Unlock()
121 o.cond.Broadcast()
122}
123
124// Step queues a step snapshot.
125func (o *outbox) Step(s *ermite.StepUpdate) {
126 o.mu.Lock()
127 if o.dead {
128 o.mu.Unlock()
129 return
130 }
131 o.events = append(o.events, event{step: s})
132 o.mu.Unlock()
133 o.cond.Broadcast()
134}
135
136// Close sends what is queued, then stops the sender. It waits at most until
137// ctx ends.
138func (o *outbox) Close(ctx context.Context) {
139 o.mu.Lock()
140 o.closing = true
141 o.mu.Unlock()
142 o.cond.Broadcast()
143 select {
144 case <-o.done:
145 case <-ctx.Done():
146 o.kill()
147 <-o.done
148 }
149}
150
151// Lines returns how many lines the server holds (as far as we know).
152func (o *outbox) Lines() int64 {
153 o.mu.Lock()
154 defer o.mu.Unlock()
155 return o.offset
156}
157
158// next waits for a batch of events. It returns nil when the outbox is done.
159func (o *outbox) next(ctx context.Context) []event {
160 o.mu.Lock()
161 defer o.mu.Unlock()
162 for len(o.events) == 0 {
163 if o.dead || o.closing {
164 return nil
165 }
166 o.cond.Wait()
167 }
168 // Let a burst of lines fill the batch, unless it is full, a step event is
169 // at the front, or we are closing.
170 if o.events[0].line != nil && !o.closing && o.queued < maxBatchBytes && len(o.events) < maxBatchLines {
171 o.mu.Unlock()
172 sleep(ctx, o.t.LogFlush)
173 o.mu.Lock()
174 if o.dead {
175 return nil
176 }
177 }
178 if len(o.events) == 0 {
179 return nil
180 }
181 first := o.events[0]
182 if first.step != nil {
183 o.events = o.events[1:]
184 return []event{first}
185 }
186 // A run of lines of the same step, within the limits.
187 n, bytes := 0, 0
188 step := first.line.step
189 for _, e := range o.events {
190 if e.line == nil || e.line.step != step || n == maxBatchLines || (n > 0 && bytes+len(e.line.text) > maxBatchBytes) {
191 break
192 }
193 n++
194 bytes += len(e.line.text)
195 }
196 batch := append([]event(nil), o.events[:n]...)
197 o.events = o.events[n:]
198 for _, e := range batch {
199 o.queued -= len(e.line.text) + 32
200 }
201 o.cond.Broadcast() // producers waiting for room
202 return batch
203}
204
205func (o *outbox) loop(ctx context.Context) {
206 for {
207 batch := o.next(ctx)
208 if batch == nil {
209 return
210 }
211 if batch[0].step != nil {
212 o.sendStep(ctx, batch[0].step)
213 } else {
214 o.sendLines(ctx, batch)
215 }
216 o.mu.Lock()
217 dead := o.dead
218 o.mu.Unlock()
219 if dead {
220 return
221 }
222 }
223}
224
225// retry runs call until it succeeds, the server gives a verdict, or ctx ends.
226// It returns nil on success.
227func (o *outbox) retry(ctx context.Context, what string, call func(context.Context) error) error {
228 b := backoff{min: o.t.BackoffMin, max: o.t.BackoffMax}
229 for {
230 cctx, cancel := context.WithTimeout(ctx, o.t.CallTimeout)
231 err := call(cctx)
232 cancel()
233 if err == nil {
234 return nil
235 }
236 if ctx.Err() != nil {
237 return ctx.Err()
238 }
239 switch classify(err) {
240 case kindEnded, kindAbandoned:
241 o.onEnded(err)
242 o.kill()
243 return err
244 case kindPermanent:
245 return err
246 }
247 o.log.Warn().Err(err).Str("call", what).Msg("call failed, retrying")
248 if !sleep(ctx, b.next()) {
249 return ctx.Err()
250 }
251 }
252}
253
254func (o *outbox) sendStep(ctx context.Context, s *ermite.StepUpdate) {
255 err := o.retry(ctx, "UpdateJob", func(c context.Context) error {
256 _, err := o.client.UpdateJob(c, connect.NewRequest(&ermite.UpdateJobRequest{JobId: o.jobID, Steps: []*ermite.StepUpdate{s}}))
257 return err
258 })
259 if err != nil && ctx.Err() == nil && classify(err) == kindPermanent {
260 o.log.Error().Err(err).Int32("step", s.Number).Msg("UpdateJob refused")
261 }
262}
263
264func (o *outbox) sendLines(ctx context.Context, batch []event) {
265 req := &ermite.AppendLogRequest{JobId: o.jobID, Step: batch[0].line.step}
266 for _, e := range batch {
267 req.Lines = append(req.Lines, &ermite.LogLine{Time: timestamppb.New(time.Unix(0, e.line.unixNano)), Text: e.line.text, Continued: e.line.continued})
268 }
269 // The offset is fixed before the first attempt: every retry sends the same
270 // lines at the same offset.
271 req.Offset = o.offset
272 var next int64
273 err := o.retry(ctx, "AppendLog", func(c context.Context) error {
274 resp, err := o.client.AppendLog(c, connect.NewRequest(req))
275 if err != nil {
276 return err
277 }
278 next = resp.Msg.NextOffset
279 return nil
280 })
281 switch {
282 case err == nil:
283 o.mu.Lock()
284 o.offsetMaybeStale = false
285 // The server says how many lines it holds: that is the truth. A
286 // difference means lines were stored that we did not count (a retry
287 // whose answer we lost) or dropped; continue from the server's count,
288 // so that the next offset is never a gap (out_of_range) and never a
289 // repeat of lines that would be skipped.
290 if want := req.Offset + int64(len(req.Lines)); next != want {
291 o.log.Warn().Int64("server_lines", next).Int64("expected", want).Msg("AppendLog: the server holds another number of lines, continuing from its count")
292 }
293 o.offset = next
294 o.mu.Unlock()
295 case ctx.Err() != nil:
296 // Cut short in the middle of a call or of its retries: the server may
297 // hold these lines without our having seen the answer.
298 o.mu.Lock()
299 o.offsetMaybeStale = true
300 o.mu.Unlock()
301 case classify(err) == kindPermanent:
302 // The server refused these lines; they are dropped and the offset stays
303 // where the server is.
304 o.log.Error().Err(err).Int("lines", len(req.Lines)).Int32("step", req.Step).Msg("AppendLog refused, the lines are dropped")
305 }
306}
307
308// AppendNow sends one line synchronously, after Close: the sender has stopped,
309// so nothing else uses the offset. It is for the last words of a job (an error
310// that explains why the result could not be reported).
311//
312// If Close gave up in the middle of a send, the offset may be behind the
313// server's: the line is sent anyway (the server answers with its own count, and
314// an offset that is too low makes it skip the line as a repeat or refuse it),
315// and the warning says why it may be missing.
316func (o *outbox) AppendNow(ctx context.Context, l logLine) {
317 o.mu.Lock()
318 stale := o.offsetMaybeStale
319 offset := o.offset
320 o.mu.Unlock()
321 if stale {
322 o.log.Warn().Int64("offset", offset).Msg("the log offset may be stale: Close was cut short in the middle of a send, so the last line may be skipped or refused by the server")
323 }
324 o.sendLines(ctx, []event{{line: &l}})
325}