alice.pds.demo.boltless.dev
This repository has no description
1package engine
2
3import (
4 "bufio"
5 "bytes"
6 "context"
7 "crypto/sha256"
8 "database/sql"
9 "encoding/hex"
10 "errors"
11 "fmt"
12 "io"
13 "log/slog"
14 "os/exec"
15 "strings"
16 "sync"
17 "time"
18
19 "github.com/google/uuid"
20 "tangled.org/core/spindle/db"
21
22 "tangled.org/core/spindle/models"
23 "tangled.org/core/spindle/storage"
24)
25
26// cache log steps live below the setup step (-1)
27const (
28 CacheRestoreStepIdx = -2
29 CacheSaveStepIdx = -3
30)
31
32type cacheStep struct {
33 name string
34 command string
35}
36
37func (s cacheStep) Name() string { return s.name }
38func (s cacheStep) Command() string { return s.command }
39func (s cacheStep) Kind() models.StepKind { return models.StepKindSystem }
40
41var (
42 CacheRestoreStep models.Step = cacheStep{name: "restore cache", command: "restore cached paths"}
43 CacheSaveStep models.Step = cacheStep{name: "save cache", command: "persist changed paths"}
44)
45
46type CacheRunner interface {
47 RestoreCache(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, store storage.Storage, caches []ResolvedCache, wfLogger models.WorkflowLogger) error
48 SaveCache(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, store storage.Storage, caches []ResolvedCache, wfLogger models.WorkflowLogger) error
49}
50
51type ResolvedCache struct {
52 Paths []string
53 Key string
54 Hash string
55 SaveKey string
56 RestoreID string
57 RestoreKey string
58 RestoreName string
59 CompressionLevel int
60 When string
61}
62
63func (rc ResolvedCache) saveOn(failed bool) bool {
64 return rc.When == "always" || !failed
65}
66
67const CacheExitNoPaths = 42
68
69// avoids storing an empty archive when zstd is missing
70const CacheExitNoCompressor = 43
71
72func AbsolutizePaths(paths []string, workspaceRoot string) []string {
73 abs := make([]string, 0, len(paths))
74 for _, p := range paths {
75 if !strings.HasPrefix(p, "/") {
76 p = workspaceRoot + "/" + p
77 }
78 abs = append(abs, p)
79 }
80 return abs
81}
82
83// run tar from / so absolute paths survive extraction
84// tar exits nonzero on missing paths, so only existing ones reach it
85func CacheSaveScript(paths []string, workspaceRoot string, compressionLevel int) string {
86 trimmed := make([]string, 0, len(paths))
87 for _, p := range AbsolutizePaths(paths, workspaceRoot) {
88 trimmed = append(trimmed, strings.TrimPrefix(p, "/"))
89 }
90 tail := fmt.Sprintf(`tar -cf - -C / "$@" | %s`, CacheCompressCmd(compressionLevel))
91 return fmt.Sprintf(`set -o pipefail
92command -v zstd >/dev/null 2>&1 || { echo "zstd not found in image; cannot save cache" >&2; exit %d; }
93set --
94for p in %s; do [ -e "/$p" ] && set -- "$@" "$p"; done
95if [ $# -eq 0 ]; then echo "no cache paths exist; skipping" >&2; exit %d; fi
96%s`, CacheExitNoCompressor, strings.Join(trimmed, " "), CacheExitNoPaths, tail)
97}
98
99func CacheCompressCmd(level int) string {
100 if level == 0 {
101 return "zstd -T0 -5"
102 }
103 return fmt.Sprintf("zstd -T0 -%d", level)
104}
105
106// old entries might still be gzip, so detect them instead of trusting config
107func CacheDecompressCmd(br *bufio.Reader) string {
108 head, _ := br.Peek(4)
109 if bytes.HasPrefix(head, []byte{0x1f, 0x8b}) {
110 return "gzip -dc"
111 }
112 return "zstd -dc"
113}
114
115// Put keeps draining the pipe after it returns, so the guest writer never
116// blocks on a full pipe.
117type CacheUpload struct {
118 Writer *io.PipeWriter
119 done chan error
120}
121
122func NewCacheUpload(ctx context.Context, store storage.Storage, key string) *CacheUpload {
123 pr, pw := io.Pipe()
124 u := &CacheUpload{Writer: pw, done: make(chan error, 1)}
125 go func() {
126 err := store.Put(ctx, key, pr)
127 _, _ = io.Copy(io.Discard, pr)
128 u.done <- err
129 }()
130 return u
131}
132
133func (u *CacheUpload) Abort(err error) {
134 u.Writer.CloseWithError(err)
135 <-u.done
136}
137
138// storage treats EOF as a complete archive, so only a clean exec may Finish
139func (u *CacheUpload) Finish() error {
140 u.Writer.Close()
141 return <-u.done
142}
143
144type indexedCacheStore struct {
145 storage.Storage
146 index *db.DB
147 logger *slog.Logger
148 mu sync.Mutex
149 entries map[string]string // storage key -> cache entry id
150}
151
152func (s *indexedCacheStore) Get(ctx context.Context, key string) (io.ReadCloser, error) {
153 s.mu.Lock()
154 id, ok := s.entries[key]
155 s.mu.Unlock()
156 r, err := s.Storage.Get(ctx, key)
157 if err != nil {
158 if ok && errors.Is(err, storage.ErrNotExist) {
159 _ = s.index.DeleteCacheEntry(context.WithoutCancel(ctx), id)
160 }
161 return nil, err
162 }
163 if ok {
164 if err := s.index.TouchCacheEntry(ctx, id, time.Now()); err != nil {
165 s.logger.Warn("cache usage update failed", "id", id, "err", err)
166 }
167 }
168 return r, nil
169}
170
171func (s *indexedCacheStore) Put(ctx context.Context, key string, r io.Reader) error {
172 s.mu.Lock()
173 id, ok := s.entries[key]
174 s.mu.Unlock()
175 if !ok {
176 return fmt.Errorf("cache metadata missing for %q", key)
177 }
178 counted := &countingReader{r: r}
179 if err := s.Storage.Put(ctx, key, counted); err != nil {
180 return err
181 }
182 superseded, err := s.index.MarkCacheEntryReady(ctx, id, counted.n, time.Now())
183 if err != nil {
184 _ = s.Storage.Delete(context.WithoutCancel(ctx), key)
185 return fmt.Errorf("mark cache ready: %w", err)
186 }
187 // completed saves leave the pending map so cleanup keeps their object
188 s.mu.Lock()
189 delete(s.entries, key)
190 s.mu.Unlock()
191 cleanupCtx := context.WithoutCancel(ctx)
192 for _, old := range superseded {
193 s.deleteEntry(cleanupCtx, old.StorageKey, old.ID, "replaced")
194 }
195 return nil
196}
197
198func (s *indexedCacheStore) cleanup(ctx context.Context) {
199 s.mu.Lock()
200 defer s.mu.Unlock()
201 for key, id := range s.entries {
202 s.deleteEntry(ctx, key, id, "incomplete")
203 }
204}
205
206func (s *indexedCacheStore) deleteEntry(ctx context.Context, key, id, reason string) {
207 if err := s.Storage.Delete(ctx, key); err != nil {
208 s.logger.Warn("delete "+reason+" cache failed", "key", key, "err", err)
209 return
210 }
211 if err := s.index.DeleteCacheEntry(ctx, id); err != nil {
212 s.logger.Warn("delete "+reason+" cache metadata failed", "id", id, "err", err)
213 }
214}
215
216type countingReader struct {
217 r io.Reader
218 n int64
219}
220
221func (r *countingReader) Read(p []byte) (int, error) {
222 n, err := r.r.Read(p)
223 r.n += int64(n)
224 return n, err
225}
226
227func cacheStoreForRestore(base storage.Storage, index *db.DB, l *slog.Logger, entries []ResolvedCache) storage.Storage {
228 mapped := make(map[string]string, len(entries))
229 for _, entry := range entries {
230 if entry.RestoreKey != "" {
231 mapped[entry.RestoreKey] = entry.RestoreID
232 }
233 }
234 return &indexedCacheStore{Storage: base, index: index, logger: l, entries: mapped}
235}
236
237func prepareCacheSaves(ctx context.Context, base storage.Storage, index *db.DB, l *slog.Logger, ownerDID, repoDID, engineName string, entries []ResolvedCache) (*indexedCacheStore, error) {
238 saveStore := &indexedCacheStore{
239 Storage: base,
240 index: index,
241 logger: l,
242 entries: make(map[string]string, len(entries)),
243 }
244 now := time.Now()
245 for i := range entries {
246 id := uuid.NewString()
247 entries[i].SaveKey = "objects/" + id
248 if err := index.InsertCacheEntry(ctx, db.CacheEntry{
249 ID: id,
250 StorageKey: entries[i].SaveKey,
251 OwnerDID: ownerDID,
252 RepoDID: repoDID,
253 Engine: engineName,
254 CacheKey: entries[i].Key,
255 CacheHash: entries[i].Hash,
256 State: "pending",
257 CreatedAt: now,
258 LastUsedAt: now,
259 }); err != nil {
260 saveStore.cleanup(context.WithoutCancel(ctx))
261 return nil, err
262 }
263 saveStore.entries[entries[i].SaveKey] = id
264 }
265 return saveStore, nil
266}
267
268// on a hash miss, the newest older generation still warms the build
269// unusable entries degrade to a plain miss
270func ResolveCaches(ctx context.Context, l *slog.Logger, index *db.DB, repoDID, engine, repoPath, rev string, entries []models.CacheEntry) []ResolvedCache {
271 resolved := make([]ResolvedCache, 0, len(entries))
272 for _, entry := range entries {
273 hash := ""
274 if len(entry.Hash) > 0 && repoPath != "" {
275 sum, missing := hashKeyFiles(ctx, repoPath, rev, entry.Hash)
276 for _, m := range missing {
277 l.Warn("cache hash file not in repo", "key", entry.Key, "path", m)
278 }
279 hash = sum
280 }
281 rc := ResolvedCache{
282 Paths: entry.Paths,
283 Key: entry.Key,
284 Hash: hash,
285 CompressionLevel: entry.CompressionLevel,
286 When: entry.When,
287 }
288
289 found, err := index.FindCacheEntry(ctx, repoDID, engine, entry.Key, hash)
290 if errors.Is(err, sql.ErrNoRows) && hash != "" {
291 found, err = index.FindFallbackCacheEntry(ctx, repoDID, engine, entry.Key, hash)
292 if err == nil {
293 rc.RestoreName = entry.Key + "-" + found.CacheHash
294 }
295 }
296 if err != nil && !errors.Is(err, sql.ErrNoRows) {
297 l.Warn("cache lookup failed; entry will save but not restore", "key", entry.Key, "err", err)
298 } else if err == nil {
299 rc.RestoreID = found.ID
300 rc.RestoreKey = found.StorageKey
301 }
302 resolved = append(resolved, rc)
303 }
304 return resolved
305}
306
307func hashKeyFiles(ctx context.Context, repoPath, rev string, paths []string) (string, []string) {
308 h := sha256.New()
309 var missing []string
310 hashed := 0
311 for _, p := range paths {
312 blob, err := gitBlobId(ctx, repoPath, rev, p)
313 if err != nil {
314 missing = append(missing, p)
315 continue
316 }
317 fmt.Fprintf(h, "%s=%s\n", p, blob)
318 hashed++
319 }
320 if hashed == 0 {
321 return "", missing
322 }
323 return hex.EncodeToString(h.Sum(nil))[:12], missing
324}
325
326// sparse checkouts might not have the file, but the object database does
327func gitBlobId(ctx context.Context, repoPath, rev, path string) (string, error) {
328 out, err := exec.CommandContext(ctx, "git", "-C", repoPath, "rev-parse", rev+":"+path).Output()
329 if err != nil {
330 return "", err
331 }
332 return strings.TrimSpace(string(out)), nil
333}