package main
import (
"encoding/json"
"errors"
"os"
"path/filepath"
"sort"
"craftdesign.group/quarry/internal/ci"
)
// Log retention (RFC 0009 §logs, as amended for dagnats#624's scope):
// the engine owns the hot lane; a forge that needs a verdict to stay
// explainable for years stores the logs next to the verdict. On a
// run's terminal transition the poller drains every step's chunks into
// one object per run under the derived-data directory. An engine that
// predates the lane simply leaves nothing to drain.
// ciLogPath is the drained object: data/<repo>/ci-logs/<runID>.json.
func ciLogPath(dir, repo, runID string) string {
return filepath.Join(dir, repo, "ci-logs", runID+".json")
}
// drainLogs pulls a finished run's chunks and stores them, each chunk
// stamped with its step and attempt. Every attempt is read by number:
// the engine's default is the latest, and a retried step's failing
// first attempt is exactly the log a reader needs (#269).
//
// Draining merges into what the record already holds. An attempt that
// ended (its last chunk is a terminal marker) is never replaced — the
// hot lane expiring, or a later drain racing a retry, cannot overwrite
// it. A still-running attempt is replaced only by a read that is at
// least as long. An engine without the lane is not an error, it is an
// empty drain.
func drainLogs(e *ci.Engine, dir, repo, runID string, steps map[string]ci.Step) error {
type try struct {
step string
attempt int
}
held := map[try][]ci.LogChunk{}
if prev, ok := (ciLogStore{dir: dir}).DrainedLogs(repo, runID); ok {
for _, c := range prev {
if c.Attempt == 0 {
// Drained before attempts were kept: the engine's
// default read, which is the latest attempt.
c.Attempt = max(1, steps[c.Step].Attempts)
}
k := try{c.Step, c.Attempt}
held[k] = append(held[k], c)
}
}
for step, st := range steps {
for n := 1; n <= max(1, st.Attempts); n++ {
k := try{step, n}
if attemptEnded(held[k]) {
continue
}
chunks, err := e.Logs(runID, step, n, 0, false)
if errors.Is(err, ci.ErrNoLogs) {
return nil
}
if err != nil {
return err
}
if len(chunks) == 0 || len(chunks) < len(held[k]) {
continue
}
for i := range chunks {
chunks[i].Step, chunks[i].Attempt = step, n
}
held[k] = chunks
}
}
if len(held) == 0 {
return nil
}
keys := make([]try, 0, len(held))
for k := range held {
keys = append(keys, k)
}
sort.Slice(keys, func(i, j int) bool {
if keys[i].step != keys[j].step {
return keys[i].step < keys[j].step
}
return keys[i].attempt < keys[j].attempt
})
var all []ci.LogChunk
for _, k := range keys {
all = append(all, held[k]...)
}
p := ciLogPath(dir, repo, runID)
if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
return err
}
b, err := json.Marshal(all)
if err != nil {
return err
}
return os.WriteFile(p, b, 0o644)
}
// attemptEnded reports an attempt whose last chunk is one of the
// markers dagnats writes as the final chunk of an attempt (#624):
// nothing more will ever land on it.
func attemptEnded(chunks []ci.LogChunk) bool {
if len(chunks) == 0 {
return false
}
last := chunks[len(chunks)-1]
if last.Stream != "marker" {
return false
}
switch last.Data {
case "completed", "failed", "continued", "paused":
return true
}
return false
}
// drainedLogs is the read side: the stored chunks, or nothing.
type ciLogStore struct{ dir string }
func (s ciLogStore) DrainedLogs(repo, runID string) ([]ci.LogChunk, bool) {
b, err := os.ReadFile(ciLogPath(s.dir, repo, runID))
if err != nil {
return nil, false
}
var chunks []ci.LogChunk
if json.Unmarshal(b, &chunks) != nil {
return nil, false
}
return chunks, true
}