1// SPDX-License-Identifier: AGPL-3.0-or-later
2
3// Package cilogs implements ports.CILogStore with append-only files under
4// $ERMITE_DATA_DIR/ci/logs/<run id>/<job id>/<attempt>.log.
5//
6// A file holds one line per log line, in GitHub's raw log shape: an RFC 3339
7// timestamp with nanoseconds, a space, then the text, behind a one-character
8// marker: "-" for a line, "+" for a piece that continues the line before it
9// (it keeps its own line number):
10//
11// - 2026-10-12T10:00:00.123456789Z text
12// - 2026-10-12T10:00:00.123456789Z more text of the same line
13//
14// The file is the truth for bytes and line counts: the metadata store is never
15// written per chunk. Counts are cached per open log and rebuilt from the file
16// after a restart.
17package cilogs
18
19import (
20 "bufio"
21 "bytes"
22 "context"
23 "errors"
24 "fmt"
25 "io"
26 "io/fs"
27 "os"
28 "path/filepath"
29 "strconv"
30 "sync"
31 "time"
32
33 "bnema.dev/ermite/internal/domain"
34 "bnema.dev/ermite/internal/ports"
35)
36
37// Dir is the directory of the logs under the data directory.
38func Dir(dataDir string) string { return filepath.Join(dataDir, "ci", "logs") }
39
40// timeLayout is RFC 3339 with nanoseconds, always in UTC and with all nine
41// digits, so that every prefix has the same length.
42const timeLayout = "2006-01-02T15:04:05.000000000Z"
43
44// prefixLen is the length of "- " + timestamp + " ".
45const prefixLen = 2 + len(timeLayout) + 1
46
47// Store is the file-backed log store.
48type Store struct {
49 root string
50
51 mu sync.Mutex
52 logs map[ports.CILogRef]*logState
53}
54
55var _ ports.CILogStore = (*Store)(nil)
56
57// logState is what is known of one log file.
58type logState struct {
59 mu sync.Mutex
60 loaded bool
61 count int64
62 size int64
63 offsets []int64 // byte offset of every 1024th line: index i is line i*1024
64 changed chan struct{}
65}
66
67const offsetEvery = 1024
68
69// New returns a store rooted at dir, which need not exist yet.
70func New(dir string) *Store {
71 return &Store{root: dir, logs: make(map[ports.CILogRef]*logState)}
72}
73
74func (s *Store) path(ref ports.CILogRef) string {
75 return filepath.Join(s.root, strconv.FormatInt(ref.RunID, 10), ref.JobID, strconv.Itoa(int(ref.Attempt))+".log")
76}
77
78func (s *Store) state(ref ports.CILogRef) *logState {
79 s.mu.Lock()
80 defer s.mu.Unlock()
81 st, ok := s.logs[ref]
82 if !ok {
83 st = &logState{changed: make(chan struct{})}
84 s.logs[ref] = st
85 }
86 return st
87}
88
89// validRef rejects a reference that could escape the directory.
90func validRef(ref ports.CILogRef) bool {
91 if ref.RunID < 1 || ref.Attempt < 1 {
92 return false
93 }
94 _, _, _, ok := domain.ParseJobID(ref.JobID)
95 return ok
96}
97
98// load reads the file to learn its line count and size, once. The caller
99// holds st.mu.
100func (s *Store) load(ref ports.CILogRef, st *logState) error {
101 if st.loaded {
102 return nil
103 }
104 f, err := os.Open(s.path(ref))
105 if errors.Is(err, fs.ErrNotExist) {
106 st.loaded = true
107 return nil
108 }
109 if err != nil {
110 return fmt.Errorf("open log: %w", err)
111 }
112 defer func() { _ = f.Close() }()
113 r := bufio.NewReaderSize(f, 1<<16)
114 var count, pos int64
115 for {
116 line, err := r.ReadBytes('\n')
117 if len(line) > 0 && line[len(line)-1] == '\n' {
118 if count%offsetEvery == 0 {
119 st.offsets = append(st.offsets, pos)
120 }
121 count++
122 pos += int64(len(line))
123 } else if len(line) > 0 {
124 // A torn last line from a crash: cut it, it was never acknowledged.
125 if terr := os.Truncate(s.path(ref), pos); terr != nil {
126 return fmt.Errorf("repair log: %w", terr)
127 }
128 }
129 if err != nil {
130 if errors.Is(err, io.EOF) {
131 break
132 }
133 return fmt.Errorf("read log: %w", err)
134 }
135 }
136 st.count, st.size, st.loaded = count, pos, true
137 return nil
138}
139
140func encodeLine(l domain.CILogLine) []byte {
141 b := make([]byte, 0, prefixLen+len(l.Text)+1)
142 if l.Continued {
143 b = append(b, '+', ' ')
144 } else {
145 b = append(b, '-', ' ')
146 }
147 b = l.Time.UTC().AppendFormat(b, timeLayout)
148 b = append(b, ' ')
149 b = append(b, l.Text...)
150 return append(b, '\n')
151}
152
153func decodeLine(b []byte) (domain.CILogLine, error) {
154 b = bytes.TrimSuffix(b, []byte{'\n'})
155 if len(b) < prefixLen || (b[0] != '-' && b[0] != '+') {
156 return domain.CILogLine{}, errors.New("malformed log line")
157 }
158 t, err := time.Parse(timeLayout, string(b[2:2+len(timeLayout)]))
159 if err != nil {
160 return domain.CILogLine{}, fmt.Errorf("malformed log time: %w", err)
161 }
162 return domain.CILogLine{Time: t, Text: string(b[prefixLen:]), Continued: b[0] == '+'}, nil
163}
164
165// Append appends lines at offset.
166func (s *Store) Append(ctx context.Context, ref ports.CILogRef, offset int64, lines []domain.CILogLine, limit int64) (int64, int, error) {
167 return s.append(ctx, ref, offset, lines, limit, false)
168}
169
170// AppendForced appends lines whatever the limit.
171func (s *Store) AppendForced(ctx context.Context, ref ports.CILogRef, lines []domain.CILogLine) (int64, error) {
172 next, _, err := s.append(ctx, ref, -1, lines, 0, true)
173 return next, err
174}
175
176func (s *Store) append(_ context.Context, ref ports.CILogRef, offset int64, lines []domain.CILogLine, limit int64, force bool) (int64, int, error) {
177 if !validRef(ref) {
178 return 0, 0, domain.ErrNotFound
179 }
180 st := s.state(ref)
181 st.mu.Lock()
182 defer st.mu.Unlock()
183 if err := s.load(ref, st); err != nil {
184 return 0, 0, err
185 }
186 if offset < 0 {
187 offset = st.count
188 }
189 if offset > st.count {
190 return 0, 0, &ports.ErrCILogGap{Count: st.count}
191 }
192 skip := st.count - offset
193 if skip >= int64(len(lines)) {
194 return st.count, 0, nil
195 }
196 fresh := lines[skip:]
197 var buf bytes.Buffer
198 for _, l := range fresh {
199 buf.Write(encodeLine(l))
200 }
201 if !force && st.size+int64(buf.Len()) > limit {
202 return st.count, 0, ports.ErrCILogLimit
203 }
204 if err := os.MkdirAll(filepath.Dir(s.path(ref)), 0o700); err != nil {
205 return 0, 0, fmt.Errorf("create log directory: %w", err)
206 }
207 f, err := os.OpenFile(s.path(ref), os.O_WRONLY|os.O_APPEND|os.O_CREATE, 0o600)
208 if err != nil {
209 return 0, 0, fmt.Errorf("open log: %w", err)
210 }
211 n, werr := f.Write(buf.Bytes())
212 cerr := f.Close()
213 if werr != nil || cerr != nil {
214 // Cut what was written so that the file keeps whole lines.
215 _ = os.Truncate(s.path(ref), st.size)
216 return 0, 0, fmt.Errorf("write log: %w", errors.Join(werr, cerr))
217 }
218 pos := st.size
219 for i, l := range fresh {
220 if (st.count+int64(i))%offsetEvery == 0 {
221 st.offsets = append(st.offsets, pos)
222 }
223 pos += int64(len(encodeLine(l)))
224 }
225 st.count += int64(len(fresh))
226 st.size += int64(n)
227 close(st.changed)
228 st.changed = make(chan struct{})
229 return st.count, len(fresh), nil
230}
231
232// Count returns the number of lines held.
233func (s *Store) Count(_ context.Context, ref ports.CILogRef) (int64, error) {
234 if !validRef(ref) {
235 return 0, nil
236 }
237 st := s.state(ref)
238 st.mu.Lock()
239 defer st.mu.Unlock()
240 if err := s.load(ref, st); err != nil {
241 return 0, err
242 }
243 return st.count, nil
244}
245
246// Read returns lines from line number from.
247//
248//hexcheck:ignore suspicious-business-logic-in-adapter reads a bounded snapshot of stored log lines using sparse offsets and maps missing or out-of-range reads to port errors
249func (s *Store) Read(_ context.Context, ref ports.CILogRef, from int64, maxLines, maxBytes int) ([]domain.CILogLine, error) {
250 if !validRef(ref) || from < 0 {
251 return nil, domain.ErrNotFound
252 }
253 st := s.state(ref)
254 st.mu.Lock()
255 if err := s.load(ref, st); err != nil {
256 st.mu.Unlock()
257 return nil, err
258 }
259 count, size := st.count, st.size
260 var start int64
261 var first int64
262 if idx := from / offsetEvery; idx < int64(len(st.offsets)) {
263 start, first = st.offsets[idx], idx*offsetEvery
264 }
265 st.mu.Unlock()
266 if from > count {
267 return nil, domain.ErrOutOfRange
268 }
269 if _, err := os.Stat(s.path(ref)); errors.Is(err, fs.ErrNotExist) {
270 if from == 0 && count == 0 {
271 return nil, domain.ErrNotFound
272 }
273 return nil, domain.ErrNotFound
274 }
275 if from == count {
276 return nil, nil
277 }
278 f, err := os.Open(s.path(ref))
279 if err != nil {
280 return nil, fmt.Errorf("open log: %w", err)
281 }
282 defer func() { _ = f.Close() }()
283 // Never read past the size seen under the lock: a concurrent append is
284 // for the next call.
285 r := bufio.NewReaderSize(io.NewSectionReader(f, start, size-start), 1<<16)
286 var out []domain.CILogLine
287 total := 0
288 for n := first; n < count; n++ {
289 raw, err := r.ReadBytes('\n')
290 if err != nil {
291 return nil, fmt.Errorf("read log: %w", err)
292 }
293 if n < from {
294 continue
295 }
296 l, err := decodeLine(raw)
297 if err != nil {
298 return nil, fmt.Errorf("log line %d: %w", n, err)
299 }
300 if len(out) > 0 && (len(out) >= maxLines || total+len(l.Text) > maxBytes) {
301 break
302 }
303 total += len(l.Text)
304 out = append(out, l)
305 }
306 return out, nil
307}
308
309// Changed returns a channel closed at the next append.
310func (s *Store) Changed(ref ports.CILogRef) <-chan struct{} {
311 st := s.state(ref)
312 st.mu.Lock()
313 defer st.mu.Unlock()
314 return st.changed
315}
316
317// Reset empties the log.
318func (s *Store) Reset(_ context.Context, ref ports.CILogRef) error {
319 if !validRef(ref) {
320 return nil
321 }
322 st := s.state(ref)
323 st.mu.Lock()
324 defer st.mu.Unlock()
325 if err := os.Remove(s.path(ref)); err != nil && !errors.Is(err, fs.ErrNotExist) {
326 return fmt.Errorf("remove log: %w", err)
327 }
328 st.loaded, st.count, st.size, st.offsets = true, 0, 0, nil
329 close(st.changed)
330 st.changed = make(chan struct{})
331 return nil
332}
333
334// RemoveRuns deletes the logs of the runs.
335func (s *Store) RemoveRuns(_ context.Context, runIDs []int64) error {
336 var errs []error
337 for _, id := range runIDs {
338 if id < 1 {
339 continue
340 }
341 if err := os.RemoveAll(filepath.Join(s.root, strconv.FormatInt(id, 10))); err != nil {
342 errs = append(errs, err)
343 }
344 }
345 s.mu.Lock()
346 for ref := range s.logs {
347 for _, id := range runIDs {
348 if ref.RunID == id {
349 delete(s.logs, ref)
350 }
351 }
352 }
353 s.mu.Unlock()
354 return errors.Join(errs...)
355}
356
357// RunIDs lists the runs that have a log directory.
358func (s *Store) RunIDs(_ context.Context) ([]int64, error) {
359 entries, err := os.ReadDir(s.root)
360 if errors.Is(err, fs.ErrNotExist) {
361 return nil, nil
362 }
363 if err != nil {
364 return nil, fmt.Errorf("list logs: %w", err)
365 }
366 var ids []int64
367 for _, e := range entries {
368 if id, err := strconv.ParseInt(e.Name(), 10, 64); err == nil && e.IsDir() {
369 ids = append(ids, id)
370 }
371 }
372 return ids, nil
373}