package main
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"log"
"net/http"
"os"
"path/filepath"
"sort"
"strings"
"sync"
"time"
"craftdesign.group/quarry/internal/ci"
"craftdesign.group/quarry/internal/s3"
)
// Offload (#100): a drained run log is also kept in the forge's bucket,
// and the local file becomes a cache. The bucket is the forge binding
// (the empty-name row of the bucket table), so no credentials are
// configured anywhere new. With no binding the whole thing is inert.
//
// A log counts as uploaded when <runID>.json.sent holds the SHA-256 of
// the exact bytes the bucket accepted. A re-drain changes the bytes, so
// the marker stops matching and the log is neither evicted nor
// believed safe until it is sent again.
const (
ciLogSentSuffix = ".sent"
ciLogMissFor = 30 * time.Second
ciLogSweepEvery = time.Minute
ciLogMaxObject = 64 << 20
ciLogOpTimeout = 15 * time.Second
)
type ciLogOffload struct {
bucket func() (s3.Bucket, bool)
client *http.Client
maxAge time.Duration // 0: never evict by age
maxBytes int64 // 0: no size budget
mu sync.Mutex
miss map[string]time.Time
lastSweep time.Time
}
func newCILogOffload(bucket func() (s3.Bucket, bool), client *http.Client, maxAge time.Duration, maxBytes int64) *ciLogOffload {
if client == nil {
client = &http.Client{Timeout: ciLogOpTimeout}
}
return &ciLogOffload{bucket: bucket, client: client, maxAge: maxAge, maxBytes: maxBytes,
miss: map[string]time.Time{}}
}
func ciLogKey(repo, runID string) string { return "ci-logs/" + repo + "/" + runID + ".json" }
func sumHex(b []byte) string {
s := sha256.Sum256(b)
return hex.EncodeToString(s[:])
}
// fetch reads a log from the bucket. Misses and failures are
// remembered for a short while: the run page asks for every live run
// on every refresh, and a live run has nothing in the bucket yet.
func (o *ciLogOffload) fetch(repo, runID string) ([]byte, bool) {
b, ok := o.bucket()
if !ok {
return nil, false
}
key := ciLogKey(repo, runID)
o.mu.Lock()
if t, seen := o.miss[key]; seen && time.Since(t) < ciLogMissFor {
o.mu.Unlock()
return nil, false
}
o.mu.Unlock()
ctx, cancel := context.WithTimeout(context.Background(), ciLogOpTimeout)
defer cancel()
body, err := b.GetMax(ctx, o.client, key, ciLogMaxObject, time.Now().UTC())
if err != nil {
var pe *s3.ProbeError
if !errors.As(err, &pe) || pe.Status != http.StatusNotFound {
log.Printf("ci: log read-through %s: %v", key, err)
}
o.mu.Lock()
if len(o.miss) > 4096 {
clear(o.miss)
}
o.miss[key] = time.Now()
o.mu.Unlock()
return nil, false
}
return body, true
}
// Sweep uploads every drained log the bucket does not hold yet, then
// evicts uploaded ones past the age or size budget. It does nothing
// without a binding.
func (s ciLogStore) Sweep(now time.Time) {
o := s.off
if o == nil {
return
}
b, ok := o.bucket()
if !ok {
return
}
files, _ := filepath.Glob(filepath.Join(s.dir, "*", "ci-logs", "*.json"))
type held struct {
path string
size int64
mod time.Time
sent bool
}
var all []held
var total int64
for _, p := range files {
fi, err := os.Stat(p)
if err != nil {
continue
}
body, err := os.ReadFile(p)
if err != nil {
continue
}
h := sumHex(body)
sent := false
if m, err := os.ReadFile(p + ciLogSentSuffix); err == nil && strings.TrimSpace(string(m)) == h {
sent = true
}
if !sent {
repo := filepath.Base(filepath.Dir(filepath.Dir(p)))
run := strings.TrimSuffix(filepath.Base(p), ".json")
ctx, cancel := context.WithTimeout(context.Background(), ciLogOpTimeout)
err := b.Put(ctx, o.client, ciLogKey(repo, run), "application/json", body, now.UTC())
cancel()
if err != nil {
log.Printf("ci: log upload %s/%s: %v", repo, run, err)
} else if err := writeAtomic(p+ciLogSentSuffix, []byte(h)); err == nil {
sent = true
}
}
total += fi.Size()
all = append(all, held{p, fi.Size(), fi.ModTime(), sent})
}
sort.Slice(all, func(i, j int) bool { return all[i].mod.Before(all[j].mod) })
for _, h := range all {
tooOld := o.maxAge > 0 && now.Sub(h.mod) > o.maxAge
tooBig := o.maxBytes > 0 && total > o.maxBytes
if !h.sent || !(tooOld || tooBig) {
continue
}
// Re-check just before removal: a drain may have rewritten it.
body, err := os.ReadFile(h.path)
if err != nil {
continue
}
m, err := os.ReadFile(h.path + ciLogSentSuffix)
if err != nil || strings.TrimSpace(string(m)) != sumHex(body) {
continue
}
if os.Remove(h.path) == nil {
total -= h.size
_ = os.Remove(h.path + ciLogSentSuffix)
}
}
}
// Maintain is Sweep for a caller that ticks often: it runs at most
// once a minute.
func (s ciLogStore) Maintain(now time.Time) {
o := s.off
if o == nil {
return
}
o.mu.Lock()
due := o.lastSweep.IsZero() || now.Sub(o.lastSweep) >= ciLogSweepEvery
if due {
o.lastSweep = now
}
o.mu.Unlock()
if due {
s.Sweep(now)
}
}
// cache stores bucket bytes as the local copy, marked as uploaded.
func (s ciLogStore) cache(repo, runID string, body []byte) {
p := ciLogPath(s.dir, repo, runID)
if os.MkdirAll(filepath.Dir(p), 0o755) != nil {
return
}
if writeAtomic(p, body) == nil {
_ = writeAtomic(p+ciLogSentSuffix, []byte(sumHex(body)))
}
}
// remoteLogs is the read-through half of DrainedLogs.
func (s ciLogStore) remoteLogs(repo, runID string) ([]ci.LogChunk, bool) {
if s.off == nil {
return nil, false
}
body, ok := s.off.fetch(repo, runID)
if !ok {
return nil, false
}
var chunks []ci.LogChunk
if json.Unmarshal(body, &chunks) != nil {
return nil, false
}
s.cache(repo, runID, body)
return chunks, true
}