1// SPDX-License-Identifier: MIT
2
3package hosttest
4
5import (
6 "io"
7 "strings"
8 "sync"
9
10 "connectrpc.com/connect"
11
12 ermite "bnema.dev/ermite-api/gen/ermite"
13)
14
15// LogStore is the log of one job as the server of runner.proto keeps it, with
16// the rules of AppendLog (the one place where they are written down for tests):
17//
18// - a request has 1 to 1,000 lines, at most 64 KiB of text in total, no line
19// feed in a line, a time on every line, and a continued first line only if
20// it follows a line;
21// - offset is the number of lines the server holds: equal appends, below skips
22// the first (count - offset) lines of the request (a retry), above is
23// out_of_range;
24// - the step of a request is at least the step of the last stored line
25// (failed_precondition otherwise; lines skipped as already stored are not
26// checked).
27//
28// A request that breaks a rule the client must never break (the first group)
29// is reported through OnViolation as well as answered with the error.
30type LogStore struct {
31 // OnViolation, when set, receives the description of each client violation.
32 OnViolation func(string)
33
34 mu sync.Mutex
35 lines []Line
36}
37
38// Lines returns a copy of the stored lines.
39func (s *LogStore) Lines() []Line {
40 s.mu.Lock()
41 defer s.mu.Unlock()
42 return append([]Line(nil), s.lines...)
43}
44
45// Apply applies a request and returns the number of lines held afterwards.
46func (s *LogStore) Apply(r *ermite.AppendLogRequest) (int64, error) {
47 s.mu.Lock()
48 defer s.mu.Unlock()
49 bad := func(code connect.Code, why string) (int64, error) {
50 if s.OnViolation != nil {
51 s.OnViolation("AppendLog: " + why)
52 }
53 return 0, connect.NewError(code, io.EOF)
54 }
55 if r.Step < 1 {
56 return bad(connect.CodeInvalidArgument, "step below 1")
57 }
58 n, bytes := len(r.Lines), 0
59 if n == 0 || n > 1000 {
60 return bad(connect.CodeInvalidArgument, "bad number of lines")
61 }
62 have := int64(len(s.lines))
63 for i, l := range r.Lines {
64 bytes += len(l.Text)
65 if strings.Contains(l.Text, "\n") || l.Time == nil {
66 return bad(connect.CodeInvalidArgument, "invalid line")
67 }
68 if l.Continued && i == 0 && have == 0 && r.Offset == 0 {
69 return bad(connect.CodeInvalidArgument, "the first line of the job is continued")
70 }
71 }
72 if bytes > 64*1024 {
73 return bad(connect.CodeInvalidArgument, "more than 64 KiB")
74 }
75 if r.Offset > have {
76 return bad(connect.CodeOutOfRange, "a gap in the offsets")
77 }
78 skip := have - r.Offset
79 if skip < int64(n) && have > 0 && r.Step < s.lines[have-1].Step {
80 return bad(connect.CodeFailedPrecondition, "a step went back")
81 }
82 for i, l := range r.Lines {
83 if int64(i) < skip {
84 continue
85 }
86 s.lines = append(s.lines, Line{Step: r.Step, Text: l.Text, Continued: l.Continued})
87 }
88 return int64(len(s.lines)), nil
89}