+1

docker-compose.yml

Reviewed

··· 183 183 SPINDLE_MICROVM_PIPELINES_AGENT_PORT: "11240" 184 184 SPINDLE_S3_LOG_BUCKET: "" 185 185 SPINDLE_MICROVM_PIPELINES_ENABLE_CGROUPS: "false" 186 186 + SPINDLE_CACHE_BACKEND: disk 186 187 # route guest nix substitution + uploads through the local ncps cache. 187 188 # ncps re-signs on serve with cache.local's key, so the guest trusts the 188 189 # matching public key below (no signing happens in spindle itself).

+76

docs/DOCS.md

Reviewed

··· 965 965 - `TANGLED_PR_SOURCE_SHA` - The commit SHA of the source 966 966 branch 967 967 968 968 + ### Cache 969 969 + 970 970 + The `cache` field lets a workflow persist directories across 971 971 + pipeline runs. Before the first step, the engine looks up 972 972 + each entry's key and extracts the matching archive into the 973 973 + workspace; after all steps succeed, the paths are archived 974 974 + again and stored back under the key. 975 975 + 976 976 + - `key`: name this cache is saved under. Keys are scoped to 977 977 + the repository and the engine specified. 978 978 + - `hash`: **optional** list of repo files (lockfiles, 979 979 + manifests) whose content is folded into the key. The entry 980 980 + is stored as `<key>-<hash of those files>`, so editing 981 981 + `go.sum` automatically rotates the cache without bumping 982 982 + the key by hand. Paths are relative to the repository 983 983 + root and are read from git at the commit being built. 984 984 + - `paths`: paths to archive. Relative paths are anchored at 985 985 + the repository checkout, the directory steps start in 986 986 + (`/workspace/repo` on microvm, `/tangled/workspace` on 987 987 + nixery). Absolute paths work too, for caching directories 988 988 + outside the checkout, but note they name engine-specific 989 989 + locations. All paths must be writable by the CI user and 990 990 + contain no spaces. 991 991 + - `compression-level`: **optional** zstd level, `1` 992 992 + (fastest) to `19` (smallest). defaults to (`5`). 993 993 + - `when`: **optional** save policy: `on-success` (the 994 994 + default) or `always`. 995 995 + 996 996 + When the exact key (or generation, with `hash`) misses, the 997 997 + newest older generation under the same key is restored. 998 998 + 999 999 + ```yaml 1000 1000 + cache: 1001 1001 + - key: go-mod 1002 1002 + hash: 1003 1003 + - go.sum 1004 1004 + - go.mod 1005 1005 + paths: 1006 1006 + - .gocache 1007 1007 + ``` 1008 1008 + 1009 1009 + Caches are only restored and saved for trusted pipelines 1010 1010 + (pushes and same-repository pull requests). Pipelines 1011 1011 + building untrusted code, like pull requests from forks, 1012 1012 + skip the cache entirely. Saving follows each entry's `when` 1013 1013 + policy, except on timeout, when nothing is saved. A cache 1014 1014 + miss or failure never fails the workflow. 1015 1015 + 1016 1016 + The spindle operator chooses the storage backend. See 1017 1017 + [Running spindle](#running-spindle). If no backend is 1018 1018 + configured, `cache` entries are ignored. 1019 1019 + 968 1020 ### Steps 969 1021 970 1022 The `steps` field allows you to define what steps should run ··· 1509 1561 trusted public keys for those caches. 1510 1562 - `SPINDLE_NIX_CACHE_UPLOAD_URL`: Cache URL that paths built 1511 1563 in the guest are uploaded to. 1564 1564 + 1565 1565 + The generic CI cache (the workflow-level 1566 1566 + [`cache`](#cache) field) is configured via prefix 1567 1567 + `SPINDLE_CACHE_`. 1568 1568 + 1569 1569 + - `SPINDLE_CACHE_BACKEND`: Storage backend, `disk` or `s3` 1570 1570 + (default: `""`, caching disabled). 1571 1571 + - `SPINDLE_CACHE_DISK_DIR`: Directory for the `disk` backend 1572 1572 + (default: a `cache` directory next to the spindle 1573 1573 + database). 1574 1574 + - `SPINDLE_CACHE_S3_BUCKET`: Unversioned bucket for the `s3` 1575 1575 + backend. Credentials come from the standard AWS chain and 1576 1576 + need `s3:GetBucketVersioning` in addition to object access. 1577 1577 + - `SPINDLE_CACHE_S3_PREFIX`: Key prefix inside the bucket 1578 1578 + (default: `"spindle/cache"`). 1579 1579 + - `SPINDLE_CACHE_RETENTION`: Time since the last restore or 1580 1580 + save before an entry is deleted (default: `720h`, or 30 1581 1581 + days). Set to `0` to keep entries indefinitely. 1582 1582 + - `SPINDLE_CACHE_PRUNE_INTERVAL`: How often expired entries 1583 1583 + are deleted (default: `1h`). 1584 1584 + 1585 1585 + Cache metadata and usage are tracked in spindle's SQLite 1586 1586 + database. Storage backends contain opaque objects and are 1587 1587 + never listed during lookup or cleanup. 1512 1588 1513 1589 ### Running spindle 1514 1590

+19 -4

spindle/server.go

Reviewed

··· 42 42 "tangled.org/core/spindle/models" 43 43 "tangled.org/core/spindle/queue" 44 44 "tangled.org/core/spindle/secrets" 45 45 + "tangled.org/core/spindle/storage" 45 46 "tangled.org/core/spindle/xrpc" 46 47 "tangled.org/core/tid" 47 48 "tangled.org/core/workflow" ··· 70 71 res *idresolver.Resolver 71 72 verify repoverify.Verifier 72 73 vault secrets.Manager 74 74 + cache storage.Storage 73 75 motd []byte 74 76 motdMu sync.RWMutex 75 77 rootCtx context.Context ··· 154 156 155 157 resolver := idresolver.DefaultResolver(cfg.Server.PlcUrl) 156 158 159 159 + cacheStore, err := storage.New(ctx, cfg) 160 160 + if err != nil { 161 161 + return nil, fmt.Errorf("failed to setup cache storage: %w", err) 162 162 + } 163 163 + if cacheStore != nil { 164 164 + logger.Info("cache storage enabled", "backend", cfg.Cache.Backend) 165 165 + } 166 166 + 157 167 spindle := &Spindle{ 158 168 jc: jc, 159 169 e: e, ··· 166 176 res: resolver, 167 177 verify: repoverify.New(resolver, cfg.Server.Dev), 168 178 vault: vault, 179 179 + cache: cacheStore, 169 180 motd: defaultMotd, 170 181 rootCtx: ctx, 171 182 } ··· 224 235 cfg.Server.Tap.AdminPassword = pw 225 236 logger.Info("embedded tap: using random admin password") 226 237 } 238 238 + engine.StartCachePruner(ctx, logger, d, cacheStore, cfg.Cache.Retention, cfg.Cache.PruneInterval) 239 239 + 227 240 spindle.tap = NewTapClient(spindle) 228 241 229 242 return spindle, nil ··· 820 833 } 821 834 maps.Copy(ewf.Environment, pipelineEnv) 822 835 836 836 + ewf.Engine = w.Engine 823 837 workflows[eng] = append(workflows[eng], *ewf) 824 838 } 825 839 826 840 // enqueue pipeline 827 841 ok := s.jq.Enqueue(repoDid, queue.Job{ 828 842 Run: func() error { 829 829 - engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, s.rootCtx, &models.Pipeline{ 830 830 - RepoDid: repoDid, 831 831 - Workflows: workflows, 832 832 - TrustedSource: trustedSource, 843 843 + engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, s.cache, s.rootCtx, &models.Pipeline{ 844 844 + RepoDid: repoDid, 845 845 + Workflows: workflows, 846 846 + TrustedSource: trustedSource, 847 847 + TriggerMetadata: tpl.TriggerMetadata, 833 848 }, pipelineId) 834 849 return nil 835 850 },

+1

nix/microvm/base.nix

Reviewed

··· 258 258 gz-utils 259 259 bzip2 260 260 lz4 261 261 + zstd 261 262 p7zip 262 263 ]; 263 264 # disable default nixos packages ([perl rsync strace])

+2 -1

nix/pkgs/spindle-alpine-image.nix

Reviewed

··· 30 30 }); 31 31 # we don't include gnused, xxd etc. here because busybox has them 32 32 # we want to keep the image this image small! 33 33 - guestTools = [nix bash git curl jq]; 33 33 + # zstd is not a busybox applet, and the spindle cache saves tar|zstd 34 34 + guestTools = [nix bash git curl jq pkgsStatic.zstd]; 34 35 35 36 # run by busybox at sysinit 36 37 setupScript = writeText "spindle-setup" ''

+10

spindle/config/config.go

Reviewed

··· 61 61 LogBucket string `env:"LOG_BUCKET"` 62 62 } 63 63 64 64 + type Cache struct { 65 65 + Backend string `env:"BACKEND"` // "disk" or "s3" 66 66 + DiskDir string `env:"DISK_DIR"` 67 67 + S3Bucket string `env:"S3_BUCKET"` 68 68 + S3Prefix string `env:"S3_PREFIX, default=spindle/cache"` 69 69 + Retention time.Duration `env:"RETENTION, default=720h"` 70 70 + PruneInterval time.Duration `env:"PRUNE_INTERVAL, default=1h"` 71 71 + } 72 72 + 64 73 type MicroVMPipelines struct { 65 74 ImageDir string `env:"IMAGE_DIR"` 66 75 OverlayDir string `env:"OVERLAY_DIR, default="` // where microVM temporary disks will live ··· 99 108 MicroVMPipelines MicroVMPipelines `env:",prefix=SPINDLE_MICROVM_PIPELINES_"` 100 109 NixCache NixCache `env:",prefix=SPINDLE_NIX_CACHE_"` 101 110 S3 S3 `env:",prefix=SPINDLE_S3_"` 111 111 + Cache Cache `env:",prefix=SPINDLE_CACHE_"` 102 112 } 103 113 104 114 func Load(ctx context.Context) (*Config, error) {

+233

spindle/db/cache.go

Reviewed

··· 1 1 + package db 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "sort" 6 6 + "time" 7 7 + ) 8 8 + 9 9 + type CacheEntry struct { 10 10 + ID string 11 11 + StorageKey string 12 12 + OwnerDID string 13 13 + RepoDID string 14 14 + Engine string 15 15 + CacheKey string 16 16 + CacheHash string 17 17 + SizeBytes int64 18 18 + State string 19 19 + CreatedAt time.Time 20 20 + LastUsedAt time.Time 21 21 + } 22 22 + 23 23 + const cacheEntryColumns = ` 24 24 + id, storage_key, owner_did, repo_did, engine, cache_key, cache_hash, 25 25 + size_bytes, state, created_at, last_used_at` 26 26 + 27 27 + func (d *DB) InsertCacheEntry(ctx context.Context, entry CacheEntry) error { 28 28 + _, err := d.ExecContext(ctx, ` 29 29 + insert into cache_entries ( 30 30 + id, storage_key, owner_did, repo_did, engine, cache_key, cache_hash, 31 31 + size_bytes, state, created_at, last_used_at 32 32 + ) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`, 33 33 + entry.ID, 34 34 + entry.StorageKey, 35 35 + entry.OwnerDID, 36 36 + entry.RepoDID, 37 37 + entry.Engine, 38 38 + entry.CacheKey, 39 39 + entry.CacheHash, 40 40 + entry.SizeBytes, 41 41 + entry.State, 42 42 + entry.CreatedAt.UnixNano(), 43 43 + entry.LastUsedAt.UnixNano(), 44 44 + ) 45 45 + return err 46 46 + } 47 47 + 48 48 + func (d *DB) MarkCacheEntryReady(ctx context.Context, id string, sizeBytes int64, now time.Time) ([]CacheEntry, error) { 49 49 + tx, err := d.BeginTx(ctx, nil) 50 50 + if err != nil { 51 51 + return nil, err 52 52 + } 53 53 + defer tx.Rollback() 54 54 + 55 55 + var repoDID, engine, key, hash string 56 56 + if err := tx.QueryRowContext(ctx, ` 57 57 + update cache_entries 58 58 + set state = 'ready', size_bytes = ?, last_used_at = ? 59 59 + where id = ? and state = 'pending' 60 60 + returning repo_did, engine, cache_key, cache_hash`, 61 61 + sizeBytes, now.UnixNano(), id).Scan(&repoDID, &engine, &key, &hash); err != nil { 62 62 + return nil, err 63 63 + } 64 64 + 65 65 + rows, err := tx.QueryContext(ctx, ` 66 66 + update cache_entries 67 67 + set state = 'deleting' 68 68 + where repo_did = ? and engine = ? and cache_key = ? and cache_hash = ? 69 69 + and state = 'ready' and id <> ? 70 70 + returning `+cacheEntryColumns, repoDID, engine, key, hash, id) 71 71 + if err != nil { 72 72 + return nil, err 73 73 + } 74 74 + var superseded []CacheEntry 75 75 + for rows.Next() { 76 76 + entry, err := scanCacheEntry(rows) 77 77 + if err != nil { 78 78 + rows.Close() 79 79 + return nil, err 80 80 + } 81 81 + superseded = append(superseded, *entry) 82 82 + } 83 83 + if err := rows.Close(); err != nil { 84 84 + return nil, err 85 85 + } 86 86 + if err := rows.Err(); err != nil { 87 87 + return nil, err 88 88 + } 89 89 + sort.Slice(superseded, func(i, j int) bool { 90 90 + return superseded[i].CreatedAt.After(superseded[j].CreatedAt) 91 91 + }) 92 92 + if err := tx.Commit(); err != nil { 93 93 + return nil, err 94 94 + } 95 95 + return superseded, nil 96 96 + } 97 97 + 98 98 + func (d *DB) FindCacheEntry(ctx context.Context, repoDID, engine, key, hash string) (*CacheEntry, error) { 99 99 + return scanCacheEntry(d.QueryRowContext(ctx, ` 100 100 + select `+cacheEntryColumns+` 101 101 + from cache_entries 102 102 + where repo_did = ? and engine = ? and cache_key = ? and cache_hash = ? and state = 'ready' 103 103 + order by created_at desc 104 104 + limit 1`, repoDID, engine, key, hash)) 105 105 + } 106 106 + 107 107 + func (d *DB) FindFallbackCacheEntry(ctx context.Context, repoDID, engine, key, excludeHash string) (*CacheEntry, error) { 108 108 + return scanCacheEntry(d.QueryRowContext(ctx, ` 109 109 + select `+cacheEntryColumns+` 110 110 + from cache_entries 111 111 + where repo_did = ? and engine = ? and cache_key = ? 112 112 + and cache_hash <> ? and cache_hash <> '' and state = 'ready' 113 113 + order by created_at desc 114 114 + limit 1`, repoDID, engine, key, excludeHash)) 115 115 + } 116 116 + 117 117 + func (d *DB) TouchCacheEntry(ctx context.Context, id string, now time.Time) error { 118 118 + _, err := d.ExecContext(ctx, ` 119 119 + update cache_entries set last_used_at = ? where id = ? and state = 'ready'`, now.UnixNano(), id) 120 120 + return err 121 121 + } 122 122 + 123 123 + func (d *DB) ClaimCacheEntry(ctx context.Context, id, expectedState string, expectedLastUsed time.Time) (bool, error) { 124 124 + result, err := d.ExecContext(ctx, ` 125 125 + update cache_entries 126 126 + set state = 'deleting' 127 127 + where id = ? and state = ? and last_used_at = ?`, 128 128 + id, expectedState, expectedLastUsed.UnixNano()) 129 129 + if err != nil { 130 130 + return false, err 131 131 + } 132 132 + changed, err := result.RowsAffected() 133 133 + return changed == 1, err 134 134 + } 135 135 + 136 136 + func (d *DB) RestoreCacheEntryState(ctx context.Context, id, state string) error { 137 137 + _, err := d.ExecContext(ctx, ` 138 138 + update cache_entries set state = ? where id = ? and state = 'deleting'`, state, id) 139 139 + return err 140 140 + } 141 141 + 142 142 + func (d *DB) ExpiredCacheEntries(ctx context.Context, readyBefore, pendingBefore time.Time, limit int) ([]CacheEntry, error) { 143 143 + ready, err := d.queryCacheEntries(ctx, ` 144 144 + select `+cacheEntryColumns+` 145 145 + from cache_entries 146 146 + where state = 'ready' and last_used_at < ? 147 147 + order by last_used_at 148 148 + limit ?`, readyBefore.UnixNano(), limit) 149 149 + if err != nil { 150 150 + return nil, err 151 151 + } 152 152 + recovery, err := d.queryCacheEntries(ctx, ` 153 153 + select `+cacheEntryColumns+` 154 154 + from cache_entries 155 155 + where state in ('pending', 'deleting') and created_at < ? 156 156 + order by created_at 157 157 + limit ?`, pendingBefore.UnixNano(), limit) 158 158 + if err != nil { 159 159 + return nil, err 160 160 + } 161 161 + entries := append(ready, recovery...) 162 162 + sort.Slice(entries, func(i, j int) bool { 163 163 + left, right := entries[i].CreatedAt, entries[j].CreatedAt 164 164 + if entries[i].State == "ready" { 165 165 + left = entries[i].LastUsedAt 166 166 + } 167 167 + if entries[j].State == "ready" { 168 168 + right = entries[j].LastUsedAt 169 169 + } 170 170 + return left.Before(right) 171 171 + }) 172 172 + if len(entries) > limit { 173 173 + entries = entries[:limit] 174 174 + } 175 175 + return entries, nil 176 176 + } 177 177 + 178 178 + func (d *DB) queryCacheEntries(ctx context.Context, query string, args ...any) ([]CacheEntry, error) { 179 179 + rows, err := d.QueryContext(ctx, query, args...) 180 180 + if err != nil { 181 181 + return nil, err 182 182 + } 183 183 + defer rows.Close() 184 184 + var entries []CacheEntry 185 185 + for rows.Next() { 186 186 + entry, err := scanCacheEntry(rows) 187 187 + if err != nil { 188 188 + return nil, err 189 189 + } 190 190 + entries = append(entries, *entry) 191 191 + } 192 192 + return entries, rows.Err() 193 193 + } 194 194 + 195 195 + func (d *DB) DeleteCacheEntry(ctx context.Context, id string) error { 196 196 + _, err := d.ExecContext(ctx, `delete from cache_entries where id = ?`, id) 197 197 + return err 198 198 + } 199 199 + 200 200 + func (d *DB) CacheUsageByOwner(ctx context.Context, ownerDID string) (bytes, count int64, err error) { 201 201 + err = d.QueryRowContext(ctx, ` 202 202 + select coalesce(sum(size_bytes), 0), count(*) 203 203 + from cache_entries 204 204 + where owner_did = ? and state in ('ready', 'deleting')`, ownerDID).Scan(&bytes, &count) 205 205 + return bytes, count, err 206 206 + } 207 207 + 208 208 + type cacheEntryScanner interface { 209 209 + Scan(dest ...any) error 210 210 + } 211 211 + 212 212 + func scanCacheEntry(row cacheEntryScanner) (*CacheEntry, error) { 213 213 + var entry CacheEntry 214 214 + var createdAt, lastUsedAt int64 215 215 + if err := row.Scan( 216 216 + &entry.ID, 217 217 + &entry.StorageKey, 218 218 + &entry.OwnerDID, 219 219 + &entry.RepoDID, 220 220 + &entry.Engine, 221 221 + &entry.CacheKey, 222 222 + &entry.CacheHash, 223 223 + &entry.SizeBytes, 224 224 + &entry.State, 225 225 + &createdAt, 226 226 + &lastUsedAt, 227 227 + ); err != nil { 228 228 + return nil, err 229 229 + } 230 230 + entry.CreatedAt = time.Unix(0, createdAt) 231 231 + entry.LastUsedAt = time.Unix(0, lastUsedAt) 232 232 + return &entry, nil 233 233 + }

+302

spindle/db/cache_test.go

Reviewed

··· 1 1 + package db 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "database/sql" 6 6 + "errors" 7 7 + "testing" 8 8 + "time" 9 9 + ) 10 10 + 11 11 + func testCacheEntry(id, hash, state string, createdAt time.Time) CacheEntry { 12 12 + return CacheEntry{ 13 13 + ID: id, 14 14 + StorageKey: "objects/" + id, 15 15 + OwnerDID: "did:plc:owner", 16 16 + RepoDID: "did:plc:repo", 17 17 + Engine: "microvm", 18 18 + CacheKey: "dependencies", 19 19 + CacheHash: hash, 20 20 + SizeBytes: 10, 21 21 + State: state, 22 22 + CreatedAt: createdAt, 23 23 + LastUsedAt: createdAt, 24 24 + } 25 25 + } 26 26 + 27 27 + func insertTestCacheEntry(t *testing.T, d *DB, entry CacheEntry) { 28 28 + t.Helper() 29 29 + if err := d.InsertCacheEntry(context.Background(), entry); err != nil { 30 30 + t.Fatalf("InsertCacheEntry(%s): %v", entry.ID, err) 31 31 + } 32 32 + } 33 33 + 34 34 + func TestCacheEntryLookup(t *testing.T) { 35 35 + ctx := context.Background() 36 36 + d := newTestDB(t) 37 37 + base := time.Date(2026, 1, 2, 3, 4, 5, 6, time.UTC) 38 38 + 39 39 + insertTestCacheEntry(t, d, testCacheEntry("exact-old", "requested", "ready", base)) 40 40 + insertTestCacheEntry(t, d, testCacheEntry("exact-new", "requested", "ready", base.Add(time.Second))) 41 41 + insertTestCacheEntry(t, d, testCacheEntry("exact-pending", "requested", "pending", base.Add(2*time.Second))) 42 42 + insertTestCacheEntry(t, d, testCacheEntry("fallback-old", "old-a", "ready", base.Add(3*time.Second))) 43 43 + insertTestCacheEntry(t, d, testCacheEntry("fallback-new", "old-b", "ready", base.Add(4*time.Second))) 44 44 + insertTestCacheEntry(t, d, testCacheEntry("fallback-pending", "old-c", "pending", base.Add(5*time.Second))) 45 45 + 46 46 + exact, err := d.FindCacheEntry(ctx, "did:plc:repo", "microvm", "dependencies", "requested") 47 47 + if err != nil { 48 48 + t.Fatalf("FindCacheEntry: %v", err) 49 49 + } 50 50 + if exact.ID != "exact-new" { 51 51 + t.Fatalf("FindCacheEntry returned %q, want exact-new", exact.ID) 52 52 + } 53 53 + if !exact.CreatedAt.Equal(base.Add(time.Second)) || !exact.LastUsedAt.Equal(base.Add(time.Second)) { 54 54 + t.Fatalf("timestamps = (%v, %v), want %v", exact.CreatedAt, exact.LastUsedAt, base.Add(time.Second)) 55 55 + } 56 56 + 57 57 + fallback, err := d.FindFallbackCacheEntry(ctx, "did:plc:repo", "microvm", "dependencies", "requested") 58 58 + if err != nil { 59 59 + t.Fatalf("FindFallbackCacheEntry: %v", err) 60 60 + } 61 61 + if fallback.ID != "fallback-new" { 62 62 + t.Fatalf("FindFallbackCacheEntry returned %q, want fallback-new", fallback.ID) 63 63 + } 64 64 + 65 65 + if _, err := d.FindCacheEntry(ctx, "did:plc:repo", "microvm", "missing", "requested"); !errors.Is(err, sql.ErrNoRows) { 66 66 + t.Fatalf("missing FindCacheEntry error = %v, want sql.ErrNoRows", err) 67 67 + } 68 68 + } 69 69 + 70 70 + func TestMarkCacheEntryReadySupersedesMatchingReady(t *testing.T) { 71 71 + ctx := context.Background() 72 72 + d := newTestDB(t) 73 73 + base := time.Date(2026, 1, 3, 4, 5, 6, 7, time.UTC) 74 74 + first := testCacheEntry("first-completion", "same-hash", "pending", base) 75 75 + second := testCacheEntry("second-completion", "same-hash", "pending", base.Add(time.Second)) 76 76 + insertTestCacheEntry(t, d, first) 77 77 + insertTestCacheEntry(t, d, second) 78 78 + 79 79 + superseded, err := d.MarkCacheEntryReady(ctx, first.ID, 100, base.Add(2*time.Second)) 80 80 + if err != nil { 81 81 + t.Fatalf("mark first ready: %v", err) 82 82 + } 83 83 + if len(superseded) != 0 { 84 84 + t.Fatalf("first completion superseded %d entries, want none", len(superseded)) 85 85 + } 86 86 + superseded, err = d.MarkCacheEntryReady(ctx, second.ID, 200, base.Add(3*time.Second)) 87 87 + if err != nil { 88 88 + t.Fatalf("mark second ready: %v", err) 89 89 + } 90 90 + if len(superseded) != 1 || superseded[0].ID != first.ID || superseded[0].State != "deleting" { 91 91 + t.Fatalf("second completion superseded %+v, want deleting %s", superseded, first.ID) 92 92 + } 93 93 + 94 94 + ready, err := d.FindCacheEntry(ctx, second.RepoDID, second.Engine, second.CacheKey, second.CacheHash) 95 95 + if err != nil { 96 96 + t.Fatalf("find surviving ready entry: %v", err) 97 97 + } 98 98 + if ready.ID != second.ID || ready.SizeBytes != 200 || !ready.LastUsedAt.Equal(base.Add(3*time.Second)) { 99 99 + t.Fatalf("surviving entry = %+v, want %s (200 bytes)", ready, second.ID) 100 100 + } 101 101 + } 102 102 + 103 103 + func TestMarkCacheEntryReadyConcurrentCompletions(t *testing.T) { 104 104 + ctx := context.Background() 105 105 + d := newTestDB(t) 106 106 + base := time.Date(2026, 1, 4, 5, 6, 7, 8, time.UTC) 107 107 + first := testCacheEntry("concurrent-a", "same-hash", "pending", base) 108 108 + second := testCacheEntry("concurrent-b", "same-hash", "pending", base.Add(time.Second)) 109 109 + insertTestCacheEntry(t, d, first) 110 110 + insertTestCacheEntry(t, d, second) 111 111 + 112 112 + type result struct { 113 113 + superseded []CacheEntry 114 114 + err error 115 115 + } 116 116 + start := make(chan struct{}) 117 117 + results := make(chan result, 2) 118 118 + for _, entry := range []CacheEntry{first, second} { 119 119 + entry := entry 120 120 + go func() { 121 121 + <-start 122 122 + superseded, err := d.MarkCacheEntryReady(ctx, entry.ID, 100, base.Add(2*time.Second)) 123 123 + results <- result{superseded: superseded, err: err} 124 124 + }() 125 125 + } 126 126 + close(start) 127 127 + 128 128 + var superseded []CacheEntry 129 129 + for range 2 { 130 130 + result := <-results 131 131 + if result.err != nil { 132 132 + t.Fatalf("concurrent MarkCacheEntryReady: %v", result.err) 133 133 + } 134 134 + superseded = append(superseded, result.superseded...) 135 135 + } 136 136 + if len(superseded) != 1 || superseded[0].State != "deleting" { 137 137 + t.Fatalf("concurrent completions superseded %+v, want one deleting entry", superseded) 138 138 + } 139 139 + 140 140 + var readyCount, deletingCount int 141 141 + if err := d.QueryRowContext(ctx, ` 142 142 + select sum(state = 'ready'), sum(state = 'deleting') 143 143 + from cache_entries 144 144 + where repo_did = ? and engine = ? and cache_key = ? and cache_hash = ?`, 145 145 + first.RepoDID, first.Engine, first.CacheKey, first.CacheHash).Scan(&readyCount, &deletingCount); err != nil { 146 146 + t.Fatalf("count completion states: %v", err) 147 147 + } 148 148 + if readyCount != 1 || deletingCount != 1 { 149 149 + t.Fatalf("completion states = %d ready, %d deleting; want one each", readyCount, deletingCount) 150 150 + } 151 151 + } 152 152 + 153 153 + func TestCacheEntryReadyTouchAndExpiry(t *testing.T) { 154 154 + ctx := context.Background() 155 155 + d := newTestDB(t) 156 156 + base := time.Date(2026, 2, 3, 4, 5, 6, 7, time.UTC) 157 157 + 158 158 + readyOld := testCacheEntry("ready-old", "a", "ready", base) 159 159 + readyFresh := testCacheEntry("ready-fresh", "b", "ready", base) 160 160 + pendingOld := testCacheEntry("pending-old", "c", "pending", base) 161 161 + pendingFresh := testCacheEntry("pending-fresh", "d", "pending", base.Add(20*time.Minute)) 162 162 + deletingOld := testCacheEntry("deleting-old", "e", "deleting", base) 163 163 + deletingFresh := testCacheEntry("deleting-fresh", "f", "deleting", base.Add(20*time.Minute)) 164 164 + for _, entry := range []CacheEntry{readyOld, readyFresh, pendingOld, pendingFresh, deletingOld, deletingFresh} { 165 165 + insertTestCacheEntry(t, d, entry) 166 166 + } 167 167 + 168 168 + touchedAt := base.Add(30 * time.Minute) 169 169 + if err := d.TouchCacheEntry(ctx, readyFresh.ID, touchedAt); err != nil { 170 170 + t.Fatalf("TouchCacheEntry: %v", err) 171 171 + } 172 172 + 173 173 + expired, err := d.ExpiredCacheEntries(ctx, base.Add(10*time.Minute), base.Add(10*time.Minute), 10) 174 174 + if err != nil { 175 175 + t.Fatalf("ExpiredCacheEntries: %v", err) 176 176 + } 177 177 + got := make(map[string]bool, len(expired)) 178 178 + for _, entry := range expired { 179 179 + got[entry.ID] = true 180 180 + } 181 181 + if len(got) != 3 || !got[readyOld.ID] || !got[pendingOld.ID] || !got[deletingOld.ID] { 182 182 + t.Fatalf("expired IDs = %v, want ready-old, pending-old, and deleting-old", got) 183 183 + } 184 184 + 185 185 + limited, err := d.ExpiredCacheEntries(ctx, base.Add(10*time.Minute), base.Add(10*time.Minute), 1) 186 186 + if err != nil { 187 187 + t.Fatalf("limited ExpiredCacheEntries: %v", err) 188 188 + } 189 189 + if len(limited) != 1 { 190 190 + t.Fatalf("limited expiry count = %d, want 1", len(limited)) 191 191 + } 192 192 + } 193 193 + 194 194 + func TestCacheEntryClaim(t *testing.T) { 195 195 + ctx := context.Background() 196 196 + d := newTestDB(t) 197 197 + base := time.Date(2026, 2, 4, 5, 6, 7, 8, time.UTC) 198 198 + entry := testCacheEntry("claim-me", "hash", "ready", base) 199 199 + insertTestCacheEntry(t, d, entry) 200 200 + 201 201 + touchedAt := base.Add(time.Minute) 202 202 + if err := d.TouchCacheEntry(ctx, entry.ID, touchedAt); err != nil { 203 203 + t.Fatalf("TouchCacheEntry: %v", err) 204 204 + } 205 205 + claimed, err := d.ClaimCacheEntry(ctx, entry.ID, "ready", base) 206 206 + if err != nil { 207 207 + t.Fatalf("stale ClaimCacheEntry: %v", err) 208 208 + } 209 209 + if claimed { 210 210 + t.Fatal("stale ClaimCacheEntry claimed a touched entry") 211 211 + } 212 212 + claimed, err = d.ClaimCacheEntry(ctx, entry.ID, "ready", touchedAt) 213 213 + if err != nil { 214 214 + t.Fatalf("ClaimCacheEntry: %v", err) 215 215 + } 216 216 + if !claimed { 217 217 + t.Fatal("ClaimCacheEntry did not claim unchanged entry") 218 218 + } 219 219 + 220 220 + if err := d.TouchCacheEntry(ctx, entry.ID, base.Add(2*time.Minute)); err != nil { 221 221 + t.Fatalf("TouchCacheEntry while deleting: %v", err) 222 222 + } 223 223 + if _, err := d.MarkCacheEntryReady(ctx, entry.ID, 999, base.Add(3*time.Minute)); !errors.Is(err, sql.ErrNoRows) { 224 224 + t.Fatalf("MarkCacheEntryReady while deleting error = %v, want sql.ErrNoRows", err) 225 225 + } 226 226 + var state string 227 227 + var lastUsedAt int64 228 228 + var sizeBytes int64 229 229 + if err := d.QueryRowContext(ctx, ` 230 230 + select state, last_used_at, size_bytes from cache_entries where id = ?`, entry.ID).Scan(&state, &lastUsedAt, &sizeBytes); err != nil { 231 231 + t.Fatalf("query claimed entry: %v", err) 232 232 + } 233 233 + if state != "deleting" || lastUsedAt != touchedAt.UnixNano() || sizeBytes != entry.SizeBytes { 234 234 + t.Fatalf("claimed entry = state %q, last used %d, size %d; want deleting, %d, %d", 235 235 + state, lastUsedAt, sizeBytes, touchedAt.UnixNano(), entry.SizeBytes) 236 236 + } 237 237 + 238 238 + if err := d.RestoreCacheEntryState(ctx, entry.ID, "ready"); err != nil { 239 239 + t.Fatalf("RestoreCacheEntryState: %v", err) 240 240 + } 241 241 + restored, err := d.FindCacheEntry(ctx, entry.RepoDID, entry.Engine, entry.CacheKey, entry.CacheHash) 242 242 + if err != nil { 243 243 + t.Fatalf("find restored entry: %v", err) 244 244 + } 245 245 + if restored.State != "ready" || !restored.LastUsedAt.Equal(touchedAt) { 246 246 + t.Fatalf("restored entry = state %q, last used %v", restored.State, restored.LastUsedAt) 247 247 + } 248 248 + } 249 249 + 250 250 + func TestCacheEntryDelete(t *testing.T) { 251 251 + ctx := context.Background() 252 252 + d := newTestDB(t) 253 253 + entry := testCacheEntry("delete-me", "hash", "ready", time.Now()) 254 254 + insertTestCacheEntry(t, d, entry) 255 255 + 256 256 + if err := d.DeleteCacheEntry(ctx, entry.ID); err != nil { 257 257 + t.Fatalf("DeleteCacheEntry: %v", err) 258 258 + } 259 259 + if err := d.DeleteCacheEntry(ctx, entry.ID); err != nil { 260 260 + t.Fatalf("second DeleteCacheEntry: %v", err) 261 261 + } 262 262 + if _, err := d.FindCacheEntry(ctx, entry.RepoDID, entry.Engine, entry.CacheKey, entry.CacheHash); !errors.Is(err, sql.ErrNoRows) { 263 263 + t.Fatalf("find deleted error = %v, want sql.ErrNoRows", err) 264 264 + } 265 265 + } 266 266 + 267 267 + func TestCacheUsageByOwner(t *testing.T) { 268 268 + ctx := context.Background() 269 269 + d := newTestDB(t) 270 270 + base := time.Date(2026, 3, 4, 5, 6, 7, 8, time.UTC) 271 271 + 272 272 + first := testCacheEntry("owner-ready-a", "a", "ready", base) 273 273 + first.SizeBytes = 40 274 274 + second := testCacheEntry("owner-ready-b", "b", "ready", base) 275 275 + second.SizeBytes = 2 276 276 + deleting := testCacheEntry("owner-deleting", "old", "deleting", base) 277 277 + deleting.SizeBytes = 3 278 278 + pending := testCacheEntry("owner-pending", "c", "pending", base) 279 279 + pending.SizeBytes = 1000 280 280 + other := testCacheEntry("other-ready", "d", "ready", base) 281 281 + other.OwnerDID = "did:plc:other" 282 282 + other.SizeBytes = 500 283 283 + for _, entry := range []CacheEntry{first, second, deleting, pending, other} { 284 284 + insertTestCacheEntry(t, d, entry) 285 285 + } 286 286 + 287 287 + bytes, count, err := d.CacheUsageByOwner(ctx, "did:plc:owner") 288 288 + if err != nil { 289 289 + t.Fatalf("CacheUsageByOwner: %v", err) 290 290 + } 291 291 + if bytes != 45 || count != 3 { 292 292 + t.Fatalf("usage = (%d bytes, %d entries), want (45, 3)", bytes, count) 293 293 + } 294 294 + 295 295 + bytes, count, err = d.CacheUsageByOwner(ctx, "did:plc:missing") 296 296 + if err != nil { 297 297 + t.Fatalf("empty CacheUsageByOwner: %v", err) 298 298 + } 299 299 + if bytes != 0 || count != 0 { 300 300 + t.Fatalf("empty usage = (%d bytes, %d entries), want zero", bytes, count) 301 301 + } 302 302 + }

+25 -1

spindle/db/db.go

Reviewed

··· 107 107 updated_at text not null 108 108 ); 109 109 110 110 + create table if not exists cache_entries ( 111 111 + id text primary key, 112 112 + storage_key text unique not null, 113 113 + owner_did text not null, 114 114 + repo_did text not null, 115 115 + engine text not null, 116 116 + cache_key text not null, 117 117 + cache_hash text not null, 118 118 + size_bytes integer not null default 0, 119 119 + state text not null check (state in ('pending', 'ready', 'deleting')), 120 120 + created_at integer not null, 121 121 + last_used_at integer not null 122 122 + ); 123 123 + 124 124 + create index if not exists cache_entries_lookup 125 125 + on cache_entries (repo_did, engine, cache_key, cache_hash, created_at desc) 126 126 + where state = 'ready'; 127 127 + create index if not exists cache_entries_ready_expiry 128 128 + on cache_entries (last_used_at) where state = 'ready'; 129 129 + create index if not exists cache_entries_pending_expiry 130 130 + on cache_entries (created_at) where state in ('pending', 'deleting'); 131 131 + create index if not exists cache_entries_owner_usage 132 132 + on cache_entries (owner_did) where state in ('ready', 'deleting'); 133 133 + 110 134 create table if not exists pipelines ( 111 135 id text primary key, 112 136 repo_did text not null, ··· 136 160 return nil, err 137 161 } 138 162 139 139 - return &DB{db}, nil 163 163 + return &DB{DB: db}, nil 140 164 } 141 165 142 166 func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error {

+333

spindle/engine/cache.go

Reviewed

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

+85

spindle/engine/cache_prune.go

Reviewed

··· 1 1 + package engine 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "log/slog" 6 6 + "time" 7 7 + 8 8 + "tangled.org/core/spindle/db" 9 9 + "tangled.org/core/spindle/storage" 10 10 + ) 11 11 + 12 12 + const ( 13 13 + cachePruneBatch = 100 14 14 + cachePendingMaxAge = time.Hour 15 15 + ) 16 16 + 17 17 + func StartCachePruner(ctx context.Context, l *slog.Logger, index *db.DB, store storage.Storage, retention, interval time.Duration) { 18 18 + if store == nil || interval <= 0 { 19 19 + return 20 20 + } 21 21 + go func() { 22 22 + prune := func() { 23 23 + total := 0 24 24 + for { 25 25 + n, err := PruneCaches(ctx, index, store, time.Now(), retention, cachePendingMaxAge, cachePruneBatch) 26 26 + total += n 27 27 + if err != nil { 28 28 + l.Warn("cache prune failed", "count", total, "err", err) 29 29 + return 30 30 + } 31 31 + if n < cachePruneBatch { 32 32 + if total > 0 { 33 33 + l.Info("pruned cache entries", "count", total) 34 34 + } 35 35 + return 36 36 + } 37 37 + } 38 38 + } 39 39 + prune() 40 40 + ticker := time.NewTicker(interval) 41 41 + defer ticker.Stop() 42 42 + for { 43 43 + select { 44 44 + case <-ctx.Done(): 45 45 + return 46 46 + case <-ticker.C: 47 47 + prune() 48 48 + } 49 49 + } 50 50 + }() 51 51 + } 52 52 + 53 53 + func PruneCaches(ctx context.Context, index *db.DB, store storage.Storage, now time.Time, retention, pendingMaxAge time.Duration, limit int) (int, error) { 54 54 + readyBefore := time.Unix(0, 0) 55 55 + if retention > 0 { 56 56 + readyBefore = now.Add(-retention) 57 57 + } 58 58 + entries, err := index.ExpiredCacheEntries(ctx, readyBefore, now.Add(-pendingMaxAge), limit) 59 59 + if err != nil { 60 60 + return 0, err 61 61 + } 62 62 + pruned := 0 63 63 + for _, entry := range entries { 64 64 + if entry.State != "deleting" { 65 65 + claimed, err := index.ClaimCacheEntry(ctx, entry.ID, entry.State, entry.LastUsedAt) 66 66 + if err != nil { 67 67 + return pruned, err 68 68 + } 69 69 + if !claimed { 70 70 + continue 71 71 + } 72 72 + } 73 73 + if err := store.Delete(ctx, entry.StorageKey); err != nil { 74 74 + if entry.State != "deleting" { 75 75 + _ = index.RestoreCacheEntryState(context.WithoutCancel(ctx), entry.ID, entry.State) 76 76 + } 77 77 + return pruned, err 78 78 + } 79 79 + if err := index.DeleteCacheEntry(ctx, entry.ID); err != nil { 80 80 + return pruned, err 81 81 + } 82 82 + pruned++ 83 83 + } 84 84 + return pruned, nil 85 85 + }

+152

spindle/engine/cache_prune_test.go

Reviewed

··· 1 1 + package engine 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "errors" 6 6 + "slices" 7 7 + "testing" 8 8 + "time" 9 9 + ) 10 10 + 11 11 + func TestPruneCachesExpiryPolicies(t *testing.T) { 12 12 + ctx := context.Background() 13 13 + now := time.Date(2026, 5, 6, 7, 8, 9, 0, time.UTC) 14 14 + type spec struct { 15 15 + id string 16 16 + state string 17 17 + age time.Duration 18 18 + } 19 19 + tests := []struct { 20 20 + name string 21 21 + retention time.Duration 22 22 + pendingMax time.Duration 23 23 + entries []spec 24 24 + wantPruned int 25 25 + wantSurvive []string 26 26 + }{ 27 27 + { 28 28 + name: "ready entries expire by last use", 29 29 + retention: time.Hour, 30 30 + pendingMax: 15 * time.Minute, 31 31 + entries: []spec{{"old", "ready", 2 * time.Hour}, {"fresh", "ready", 30 * time.Minute}}, 32 32 + wantPruned: 1, 33 33 + wantSurvive: []string{"fresh"}, 34 34 + }, 35 35 + { 36 36 + name: "zero retention keeps ready entries", 37 37 + retention: 0, 38 38 + pendingMax: time.Hour, 39 39 + entries: []spec{{"ready", "ready", 24 * time.Hour}, {"pending", "pending", 2 * time.Hour}}, 40 40 + wantPruned: 1, 41 41 + wantSurvive: []string{"ready"}, 42 42 + }, 43 43 + { 44 44 + name: "pending entries expire by age", 45 45 + retention: time.Hour, 46 46 + pendingMax: time.Hour, 47 47 + entries: []spec{{"old", "pending", 2 * time.Hour}, {"fresh", "pending", 10 * time.Minute}}, 48 48 + wantPruned: 1, 49 49 + wantSurvive: []string{"fresh"}, 50 50 + }, 51 51 + } 52 52 + for _, tt := range tests { 53 53 + t.Run(tt.name, func(t *testing.T) { 54 54 + d := newCacheTestDB(t) 55 55 + store := &fakeStorage{objects: map[string][]byte{}} 56 56 + keys := map[string]string{} 57 57 + for _, e := range tt.entries { 58 58 + entry := cacheTestEntry(e.id, "did:plc:repo", "microvm", "deps", "hash", e.state, now.Add(-e.age)) 59 59 + insertCacheTestEntry(t, d, entry) 60 60 + store.objects[entry.StorageKey] = []byte(e.id) 61 61 + keys[e.id] = entry.StorageKey 62 62 + } 63 63 + 64 64 + pruned, err := PruneCaches(ctx, d, store, now, tt.retention, tt.pendingMax, 10) 65 65 + if err != nil { 66 66 + t.Fatalf("PruneCaches: %v", err) 67 67 + } 68 68 + if pruned != tt.wantPruned { 69 69 + t.Fatalf("pruned %d entries, want %d", pruned, tt.wantPruned) 70 70 + } 71 71 + for _, e := range tt.entries { 72 72 + survives := slices.Contains(tt.wantSurvive, e.id) 73 73 + if store.has(keys[e.id]) != survives || cacheTestEntryExists(t, d, e.id) != survives { 74 74 + t.Fatalf("entry %q survived = %v, want %v", e.id, !survives, survives) 75 75 + } 76 76 + } 77 77 + }) 78 78 + } 79 79 + } 80 80 + 81 81 + func TestPruneCachesSkipsEntryRefreshedAfterScan(t *testing.T) { 82 82 + ctx := context.Background() 83 83 + d := newCacheTestDB(t) 84 84 + now := time.Date(2026, 5, 6, 7, 8, 9, 0, time.UTC) 85 85 + first := cacheTestEntry("first", "did:plc:repo", "microvm", "deps", "first", "ready", now.Add(-3*time.Hour)) 86 86 + refreshed := cacheTestEntry("refreshed", "did:plc:repo", "microvm", "deps", "refreshed", "ready", now.Add(-2*time.Hour)) 87 87 + insertCacheTestEntry(t, d, first) 88 88 + insertCacheTestEntry(t, d, refreshed) 89 89 + store := &fakeStorage{objects: map[string][]byte{ 90 90 + first.StorageKey: []byte("first"), 91 91 + refreshed.StorageKey: []byte("refreshed"), 92 92 + }} 93 93 + store.onDelete = func(key string) { 94 94 + if key != first.StorageKey { 95 95 + return 96 96 + } 97 97 + if err := d.TouchCacheEntry(ctx, refreshed.ID, now); err != nil { 98 98 + t.Fatalf("TouchCacheEntry: %v", err) 99 99 + } 100 100 + } 101 101 + 102 102 + pruned, err := PruneCaches(ctx, d, store, now, time.Hour, time.Hour, 10) 103 103 + if err != nil { 104 104 + t.Fatalf("PruneCaches: %v", err) 105 105 + } 106 106 + if pruned != 1 { 107 107 + t.Fatalf("pruned %d entries, want 1", pruned) 108 108 + } 109 109 + if !store.has(refreshed.StorageKey) || !cacheTestEntryExists(t, d, refreshed.ID) { 110 110 + t.Fatal("entry refreshed after expiry scan was pruned") 111 111 + } 112 112 + } 113 113 + 114 114 + func TestPruneCachesDeleteFailureRemainsRetryable(t *testing.T) { 115 115 + for _, initialState := range []string{"ready", "deleting"} { 116 116 + t.Run(initialState, func(t *testing.T) { 117 117 + ctx := context.Background() 118 118 + d := newCacheTestDB(t) 119 119 + now := time.Date(2026, 5, 6, 7, 8, 9, 0, time.UTC) 120 120 + entry := cacheTestEntry("retry", "did:plc:repo", "microvm", "deps", "hash", initialState, now.Add(-2*time.Hour)) 121 121 + insertCacheTestEntry(t, d, entry) 122 122 + deleteErr := errors.New("delete failed") 123 123 + store := &fakeStorage{ 124 124 + objects: map[string][]byte{entry.StorageKey: []byte("archive")}, 125 125 + deleteErr: deleteErr, 126 126 + } 127 127 + 128 128 + pruned, err := PruneCaches(ctx, d, store, now, time.Hour, time.Hour, 10) 129 129 + if !errors.Is(err, deleteErr) { 130 130 + t.Fatalf("first PruneCaches error = %v, want %v", err, deleteErr) 131 131 + } 132 132 + if pruned != 0 { 133 133 + t.Fatalf("first prune count = %d, want 0", pruned) 134 134 + } 135 135 + state, _, _ := cacheTestEntryState(t, d, entry.ID) 136 136 + if state != initialState { 137 137 + t.Fatalf("state after delete failure = %q, want %q", state, initialState) 138 138 + } 139 139 + if !store.has(entry.StorageKey) { 140 140 + t.Fatal("failed delete removed object") 141 141 + } 142 142 + 143 143 + pruned, err = PruneCaches(ctx, d, store, now, time.Hour, time.Hour, 10) 144 144 + if err != nil { 145 145 + t.Fatalf("retry PruneCaches: %v", err) 146 146 + } 147 147 + if pruned != 1 || store.has(entry.StorageKey) || cacheTestEntryExists(t, d, entry.ID) { 148 148 + t.Fatalf("retry result = pruned %d, object %v, metadata %v", pruned, store.has(entry.StorageKey), cacheTestEntryExists(t, d, entry.ID)) 149 149 + } 150 150 + }) 151 151 + } 152 152 + }

+376

spindle/engine/cache_test.go

Reviewed

··· 1 1 + package engine 2 2 + 3 3 + import ( 4 4 + "bufio" 5 5 + "bytes" 6 6 + "context" 7 7 + "errors" 8 8 + "io" 9 9 + "log/slog" 10 10 + "os" 11 11 + "os/exec" 12 12 + "path/filepath" 13 13 + "slices" 14 14 + "strings" 15 15 + "testing" 16 16 + "time" 17 17 + 18 18 + "tangled.org/core/spindle/db" 19 19 + "tangled.org/core/spindle/models" 20 20 + "tangled.org/core/spindle/storage" 21 21 + ) 22 22 + 23 23 + type fakeStorage struct { 24 24 + objects map[string][]byte 25 25 + putErr error 26 26 + deleteErr error // fails once, then clears 27 27 + onDelete func(string) 28 28 + } 29 29 + 30 30 + func (f *fakeStorage) Get(_ context.Context, key string) (io.ReadCloser, error) { 31 31 + data, ok := f.objects[key] 32 32 + if !ok { 33 33 + return nil, storage.ErrNotExist 34 34 + } 35 35 + return io.NopCloser(bytes.NewReader(data)), nil 36 36 + } 37 37 + 38 38 + func (f *fakeStorage) Put(_ context.Context, key string, r io.Reader) error { 39 39 + data, err := io.ReadAll(r) 40 40 + if err != nil { 41 41 + return err 42 42 + } 43 43 + f.objects[key] = data 44 44 + return f.putErr 45 45 + } 46 46 + 47 47 + func (f *fakeStorage) Delete(_ context.Context, key string) error { 48 48 + if f.onDelete != nil { 49 49 + f.onDelete(key) 50 50 + } 51 51 + if f.deleteErr != nil { 52 52 + err := f.deleteErr 53 53 + f.deleteErr = nil 54 54 + return err 55 55 + } 56 56 + delete(f.objects, key) 57 57 + return nil 58 58 + } 59 59 + 60 60 + func (f *fakeStorage) has(key string) bool { 61 61 + _, ok := f.objects[key] 62 62 + return ok 63 63 + } 64 64 + 65 65 + func gitRepo(t *testing.T, files map[string]string) (string, string) { 66 66 + t.Helper() 67 67 + if _, err := exec.LookPath("git"); err != nil { 68 68 + t.Skip("git not available") 69 69 + } 70 70 + dir := t.TempDir() 71 71 + run := func(args ...string) { 72 72 + t.Helper() 73 73 + cmd := exec.Command("git", append([]string{"-C", dir, "-c", "user.email=t@t", "-c", "user.name=t", "-c", "init.defaultBranch=main"}, args...)...) 74 74 + if out, err := cmd.CombinedOutput(); err != nil { 75 75 + t.Fatalf("git %v: %v\n%s", args, err, out) 76 76 + } 77 77 + } 78 78 + run("init") 79 79 + for name, content := range files { 80 80 + p := filepath.Join(dir, name) 81 81 + if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { 82 82 + t.Fatal(err) 83 83 + } 84 84 + if err := os.WriteFile(p, []byte(content), 0o644); err != nil { 85 85 + t.Fatal(err) 86 86 + } 87 87 + } 88 88 + run("add", ".") 89 89 + run("commit", "-m", "init") 90 90 + return dir, "HEAD" 91 91 + } 92 92 + 93 93 + func newCacheTestDB(t *testing.T) *db.DB { 94 94 + t.Helper() 95 95 + d, err := db.Make(context.Background(), filepath.Join(t.TempDir(), "spindle.db")) 96 96 + if err != nil { 97 97 + t.Fatalf("db.Make: %v", err) 98 98 + } 99 99 + t.Cleanup(func() { _ = d.Close() }) 100 100 + return d 101 101 + } 102 102 + 103 103 + func cacheTestEntry(id, repoDID, engine, key, hash, state string, at time.Time) db.CacheEntry { 104 104 + return db.CacheEntry{ 105 105 + ID: id, 106 106 + StorageKey: "objects/" + id, 107 107 + OwnerDID: "did:plc:owner", 108 108 + RepoDID: repoDID, 109 109 + Engine: engine, 110 110 + CacheKey: key, 111 111 + CacheHash: hash, 112 112 + State: state, 113 113 + CreatedAt: at, 114 114 + LastUsedAt: at, 115 115 + } 116 116 + } 117 117 + 118 118 + func insertCacheTestEntry(t *testing.T, d *db.DB, entry db.CacheEntry) { 119 119 + t.Helper() 120 120 + if err := d.InsertCacheEntry(context.Background(), entry); err != nil { 121 121 + t.Fatalf("InsertCacheEntry(%s): %v", entry.ID, err) 122 122 + } 123 123 + } 124 124 + 125 125 + func cacheTestEntryState(t *testing.T, d *db.DB, id string) (string, int64, time.Time) { 126 126 + t.Helper() 127 127 + var state string 128 128 + var size, lastUsed int64 129 129 + if err := d.QueryRow(`select state, size_bytes, last_used_at from cache_entries where id = ?`, id).Scan(&state, &size, &lastUsed); err != nil { 130 130 + t.Fatalf("query cache entry %s: %v", id, err) 131 131 + } 132 132 + return state, size, time.Unix(0, lastUsed) 133 133 + } 134 134 + 135 135 + func cacheTestEntryExists(t *testing.T, d *db.DB, id string) bool { 136 136 + t.Helper() 137 137 + var exists bool 138 138 + if err := d.QueryRow(`select exists(select 1 from cache_entries where id = ?)`, id).Scan(&exists); err != nil { 139 139 + t.Fatalf("query cache entry existence %s: %v", id, err) 140 140 + } 141 141 + return exists 142 142 + } 143 143 + 144 144 + var discardLogger = slog.New(slog.NewTextHandler(io.Discard, nil)) 145 145 + 146 146 + func TestResolveCachesExactUnhashed(t *testing.T) { 147 147 + ctx := context.Background() 148 148 + d := newCacheTestDB(t) 149 149 + base := time.Date(2026, 4, 5, 6, 7, 8, 0, time.UTC) 150 150 + 151 151 + insertCacheTestEntry(t, d, cacheTestEntry("exact", "did:plc:repo", "microvm", "deps", "", "ready", base)) 152 152 + insertCacheTestEntry(t, d, cacheTestEntry("other-repo", "did:plc:other", "microvm", "deps", "", "ready", base.Add(time.Hour))) 153 153 + insertCacheTestEntry(t, d, cacheTestEntry("other-engine", "did:plc:repo", "nixery", "deps", "", "ready", base.Add(time.Hour))) 154 154 + insertCacheTestEntry(t, d, cacheTestEntry("pending", "did:plc:repo", "microvm", "deps", "", "pending", base.Add(time.Hour))) 155 155 + insertCacheTestEntry(t, d, cacheTestEntry("hashed-only", "did:plc:repo", "microvm", "tools", "old", "ready", base)) 156 156 + 157 157 + resolved := ResolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", "", "", []models.CacheEntry{ 158 158 + {Key: "deps", Paths: []string{"/x"}, CompressionLevel: 7, When: "always"}, 159 159 + {Key: "tools", Paths: []string{"/y"}}, 160 160 + }) 161 161 + if len(resolved) != 2 { 162 162 + t.Fatalf("got %d resolved entries, want 2", len(resolved)) 163 163 + } 164 164 + got := resolved[0] 165 165 + if got.RestoreID != "exact" || got.RestoreKey != "objects/exact" || got.RestoreName != "" { 166 166 + t.Fatalf("exact restore = (%q, %q, %q)", got.RestoreID, got.RestoreKey, got.RestoreName) 167 167 + } 168 168 + if got.SaveKey != "" { 169 169 + t.Fatalf("resolve allocated a save key %q", got.SaveKey) 170 170 + } 171 171 + if got.CompressionLevel != 7 || got.When != "always" { 172 172 + t.Fatalf("resolved metadata = %+v", got) 173 173 + } 174 174 + if resolved[1].RestoreKey != "" { 175 175 + t.Fatalf("unhashed entry used hashed fallback %q", resolved[1].RestoreKey) 176 176 + } 177 177 + } 178 178 + 179 179 + func TestResolveCachesHashExactAndFallback(t *testing.T) { 180 180 + ctx := context.Background() 181 181 + d := newCacheTestDB(t) 182 182 + repoPath, rev := gitRepo(t, map[string]string{"go.sum": "v1 contents", "go.mod": "module x"}) 183 183 + request := []models.CacheEntry{{Key: "go-mod", Hash: []string{"go.sum", "go.mod"}, Paths: []string{"/x"}}} 184 184 + 185 185 + first := ResolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", repoPath, rev, request)[0] 186 186 + if first.Hash == "" { 187 187 + t.Fatal("hash is empty") 188 188 + } 189 189 + 190 190 + base := time.Date(2026, 4, 5, 6, 7, 8, 0, time.UTC) 191 191 + insertCacheTestEntry(t, d, cacheTestEntry("exact", "did:plc:repo", "microvm", "go-mod", first.Hash, "ready", base.Add(time.Minute))) 192 192 + insertCacheTestEntry(t, d, cacheTestEntry("sibling", "did:plc:repo", "microvm", "go-modules", "newer", "ready", base.Add(time.Hour))) 193 193 + 194 194 + exact := ResolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", repoPath, rev, request)[0] 195 195 + if exact.RestoreID != "exact" || exact.RestoreKey != "objects/exact" || exact.RestoreName != "" { 196 196 + t.Fatalf("exact generation restore = %+v", exact) 197 197 + } 198 198 + 199 199 + if err := os.WriteFile(filepath.Join(repoPath, "go.sum"), []byte("v2 contents"), 0o644); err != nil { 200 200 + t.Fatal(err) 201 201 + } 202 202 + commit := exec.Command("git", "-C", repoPath, "-c", "user.email=t@t", "-c", "user.name=t", "commit", "-qam", "bump") 203 203 + if out, err := commit.CombinedOutput(); err != nil { 204 204 + t.Fatalf("commit: %v\n%s", err, out) 205 205 + } 206 206 + rotated := ResolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", repoPath, rev, request)[0] 207 207 + if rotated.Hash == first.Hash { 208 208 + t.Fatal("lockfile change did not rotate the cache hash") 209 209 + } 210 210 + if rotated.RestoreID != "exact" || rotated.RestoreKey != "objects/exact" || rotated.RestoreName != "go-mod-"+first.Hash { 211 211 + t.Fatalf("fallback restore = %+v", rotated) 212 212 + } 213 213 + } 214 214 + 215 215 + func TestResolveCachesMissingHashFiles(t *testing.T) { 216 216 + ctx := context.Background() 217 217 + d := newCacheTestDB(t) 218 218 + repoPath, rev := gitRepo(t, map[string]string{"go.mod": "module x"}) 219 219 + 220 220 + partial := ResolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", repoPath, rev, []models.CacheEntry{ 221 221 + {Key: "go-mod", Hash: []string{"go.sum", "go.mod"}, Paths: []string{"/x"}}, 222 222 + })[0] 223 223 + if len(partial.Hash) != 12 { 224 224 + t.Fatalf("partial hash = %q, want 12 characters", partial.Hash) 225 225 + } 226 226 + 227 227 + none := ResolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", repoPath, rev, []models.CacheEntry{ 228 228 + {Key: "go-mod", Hash: []string{"nope.lock"}, Paths: []string{"/x"}}, 229 229 + })[0] 230 230 + if none.Hash != "" || none.RestoreKey != "" { 231 231 + t.Fatalf("all-missing result = hash %q, restore %q", none.Hash, none.RestoreKey) 232 232 + } 233 233 + } 234 234 + 235 235 + func TestIndexedCacheStoreSaveLifecycle(t *testing.T) { 236 236 + ctx := context.Background() 237 237 + d := newCacheTestDB(t) 238 238 + old := cacheTestEntry("superseded", "did:plc:repo", "microvm", "deps", "hash", "ready", time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC)) 239 239 + insertCacheTestEntry(t, d, old) 240 240 + base := &fakeStorage{objects: map[string][]byte{old.StorageKey: []byte("old")}} 241 241 + entries := []ResolvedCache{{Key: "deps", Hash: "hash"}} 242 242 + 243 243 + store, err := prepareCacheSaves(ctx, base, d, discardLogger, "did:plc:owner", "did:plc:repo", "microvm", entries) 244 244 + if err != nil { 245 245 + t.Fatalf("prepareCacheSaves: %v", err) 246 246 + } 247 247 + saveKey := entries[0].SaveKey 248 248 + id := store.entries[saveKey] 249 249 + 250 250 + payload := []byte("cache archive bytes") 251 251 + if err := store.Put(ctx, saveKey, bytes.NewReader(payload)); err != nil { 252 252 + t.Fatalf("indexed Put: %v", err) 253 253 + } 254 254 + state, size, _ := cacheTestEntryState(t, d, id) 255 255 + if state != "ready" || size != int64(len(payload)) { 256 256 + t.Fatalf("saved metadata = state %q, size %d", state, size) 257 257 + } 258 258 + if got := base.objects[saveKey]; !bytes.Equal(got, payload) { 259 259 + t.Fatalf("stored payload = %q, want %q", got, payload) 260 260 + } 261 261 + if base.has(old.StorageKey) || cacheTestEntryExists(t, d, old.ID) { 262 262 + t.Fatal("completed save left the superseded generation") 263 263 + } 264 264 + store.cleanup(ctx) 265 265 + if !base.has(saveKey) || !cacheTestEntryExists(t, d, id) { 266 266 + t.Fatal("cleanup removed a completed cache") 267 267 + } 268 268 + } 269 269 + 270 270 + func TestIndexedCacheStoreRestoreTouchesEntry(t *testing.T) { 271 271 + ctx := context.Background() 272 272 + d := newCacheTestDB(t) 273 273 + old := time.Date(2025, 1, 2, 3, 4, 5, 0, time.UTC) 274 274 + entry := cacheTestEntry("restore", "did:plc:repo", "microvm", "deps", "hash", "ready", old) 275 275 + insertCacheTestEntry(t, d, entry) 276 276 + base := &fakeStorage{objects: map[string][]byte{entry.StorageKey: []byte("archive")}} 277 277 + store := cacheStoreForRestore(base, d, discardLogger, []ResolvedCache{{RestoreID: entry.ID, RestoreKey: entry.StorageKey}}) 278 278 + 279 279 + r, err := store.Get(ctx, entry.StorageKey) 280 280 + if err != nil { 281 281 + t.Fatalf("indexed Get: %v", err) 282 282 + } 283 283 + data, err := io.ReadAll(r) 284 284 + if err != nil { 285 285 + t.Fatalf("read restored object: %v", err) 286 286 + } 287 287 + if err := r.Close(); err != nil { 288 288 + t.Fatalf("close restored object: %v", err) 289 289 + } 290 290 + if string(data) != "archive" { 291 291 + t.Fatalf("restored payload = %q", data) 292 292 + } 293 293 + _, _, touched := cacheTestEntryState(t, d, entry.ID) 294 294 + if !touched.After(old) { 295 295 + t.Fatalf("last used = %v, want after %v", touched, old) 296 296 + } 297 297 + } 298 298 + 299 299 + func TestIndexedCacheStoreFailedPutCleanup(t *testing.T) { 300 300 + ctx := context.Background() 301 301 + d := newCacheTestDB(t) 302 302 + putErr := errors.New("put failed") 303 303 + base := &fakeStorage{objects: make(map[string][]byte), putErr: putErr} 304 304 + entries := []ResolvedCache{{Key: "deps"}} 305 305 + store, err := prepareCacheSaves(ctx, base, d, discardLogger, "did:plc:owner", "did:plc:repo", "microvm", entries) 306 306 + if err != nil { 307 307 + t.Fatalf("prepareCacheSaves: %v", err) 308 308 + } 309 309 + saveKey := entries[0].SaveKey 310 310 + 311 311 + if err := store.Put(ctx, saveKey, strings.NewReader("partial")); !errors.Is(err, putErr) { 312 312 + t.Fatalf("indexed Put error = %v, want %v", err, putErr) 313 313 + } 314 314 + store.cleanup(ctx) 315 315 + if base.has(saveKey) { 316 316 + t.Fatal("cleanup left partial object") 317 317 + } 318 318 + if cacheTestEntryExists(t, d, store.entries[saveKey]) { 319 319 + t.Fatal("cleanup left pending metadata") 320 320 + } 321 321 + } 322 322 + 323 323 + func TestIndexedCacheStoreRejectsUnpreparedKey(t *testing.T) { 324 324 + store := &indexedCacheStore{} 325 325 + if err := store.Put(context.Background(), "objects/missing", strings.NewReader("data")); err == nil { 326 326 + t.Fatal("Put accepted a key without pending metadata") 327 327 + } 328 328 + } 329 329 + 330 330 + func TestAbsolutizePaths(t *testing.T) { 331 331 + got := AbsolutizePaths([]string{"node_modules", "/root/.cache", ".gocache"}, "/workspace/repo") 332 332 + want := []string{"/workspace/repo/node_modules", "/root/.cache", "/workspace/repo/.gocache"} 333 333 + if !slices.Equal(got, want) { 334 334 + t.Fatalf("got %v, want %v", got, want) 335 335 + } 336 336 + } 337 337 + 338 338 + func TestCacheDecompressCmd(t *testing.T) { 339 339 + cases := []struct { 340 340 + name string 341 341 + head []byte 342 342 + want string 343 343 + }{ 344 344 + {"zstd", []byte{0x28, 0xb5, 0x2f, 0xfd, 0x00}, "zstd -dc"}, 345 345 + {"gzip", []byte{0x1f, 0x8b, 0x08, 0x00}, "gzip -dc"}, 346 346 + {"empty", nil, "zstd -dc"}, 347 347 + } 348 348 + for _, tc := range cases { 349 349 + got := CacheDecompressCmd(bufio.NewReader(bytes.NewReader(tc.head))) 350 350 + if got != tc.want { 351 351 + t.Errorf("%s: got %q, want %q", tc.name, got, tc.want) 352 352 + } 353 353 + } 354 354 + } 355 355 + 356 356 + func TestResolvedCacheSaveOn(t *testing.T) { 357 357 + cases := []struct { 358 358 + when string 359 359 + onFail, onPass bool 360 360 + }{ 361 361 + {"", false, true}, 362 362 + {"on-success", false, true}, 363 363 + {"always", true, true}, 364 364 + } 365 365 + for _, tc := range cases { 366 366 + rc := ResolvedCache{When: tc.when} 367 367 + if got := rc.saveOn(true); got != tc.onFail { 368 368 + t.Errorf("when=%q failed run: got %v, want %v", tc.when, got, tc.onFail) 369 369 + } 370 370 + if got := rc.saveOn(false); got != tc.onPass { 371 371 + t.Errorf("when=%q passing run: got %v, want %v", tc.when, got, tc.onPass) 372 372 + } 373 373 + } 374 374 + } 375 375 + 376 376 + var _ storage.Storage = (*fakeStorage)(nil)

+76 -2

spindle/engine/engine.go

Reviewed

··· 13 13 "tangled.org/core/spindle/db" 14 14 "tangled.org/core/spindle/models" 15 15 "tangled.org/core/spindle/secrets" 16 16 + "tangled.org/core/spindle/storage" 16 17 ) 17 18 18 19 var ( ··· 24 25 FinalizeWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error 25 26 } 26 27 27 27 - func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, db *db.DB, n *notifier.Notifier, ctx context.Context, pipeline *models.Pipeline, pipelineId models.PipelineId) { 28 28 + func StartWorkflows(l *slog.Logger, vault secrets.Manager, cfg *config.Config, db *db.DB, n *notifier.Notifier, cacheStore storage.Storage, ctx context.Context, pipeline *models.Pipeline, pipelineId models.PipelineId) { 28 29 l.Info("starting all workflows in parallel", "pipeline", pipelineId) 29 30 31 31 + isTrustedRepo := pipeline.TrustedSource && pipeline.RepoDid != "" 30 32 var allSecrets []secrets.UnlockedSecret 31 33 // never pass secrets to pipelines that run untrusted (e.g. fork) code 32 32 - if pipeline.TrustedSource && pipeline.RepoDid != "" { 34 34 + if isTrustedRepo { 33 35 if res, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(pipeline.RepoDid.String())); err == nil { 34 36 allSecrets = res 35 37 } 36 38 } else if !pipeline.TrustedSource { 37 39 l.Info("skipping secrets for untrusted pipeline source", "pipeline", pipelineId) 40 40 + } 41 41 + // untrusted runs cant read or write shared caches 42 42 + cacheOwnerDID := "" 43 43 + cacheEnabled := cacheStore != nil && isTrustedRepo 44 44 + if cacheEnabled { 45 45 + repo, err := db.GetRepoByDid(pipeline.RepoDid) 46 46 + if err != nil { 47 47 + l.Warn("cache owner lookup failed; caching disabled", "repo", pipeline.RepoDid, "err", err) 48 48 + cacheEnabled = false 49 49 + } else { 50 50 + cacheOwnerDID = repo.Owner.String() 51 51 + } 52 52 + } else if cacheStore != nil && !pipeline.TrustedSource { 53 53 + l.Info("skipping caches for untrusted pipeline source", "pipeline", pipelineId) 54 54 + } 55 55 + 56 56 + // hash the checked commit, not the working tree 57 57 + cacheRepoPath, cacheRev := "", "" 58 58 + if tm := pipeline.TriggerMetadata; tm != nil { 59 59 + if rev, err := models.ExtractCommitSHA(*tm); err == nil { 60 60 + did := pipeline.RepoDid.String() 61 61 + if tm.SourceRepo != nil && *tm.SourceRepo != "" { 62 62 + did = *tm.SourceRepo 63 63 + } 64 64 + cacheRepoPath, cacheRev = filepath.Join(cfg.Server.RepoDir, did), rev 65 65 + } else { 66 66 + l.Warn("cannot resolve pipeline commit; cache hashing disabled", "err", err) 67 67 + } 38 68 } 39 69 40 70 secretValues := make([]string, len(allSecrets)) ··· 50 80 var wg sync.WaitGroup 51 81 for eng, wfs := range pipeline.Workflows { 52 82 workflowTimeout := eng.WorkflowTimeout() 83 83 + cacheRunner, cachesSupported := eng.(CacheRunner) 53 84 l.Info("using workflow timeout", "timeout", workflowTimeout) 54 85 55 86 for _, w := range wfs { ··· 121 152 ctx, cancel := context.WithTimeout(ctx, workflowTimeout) 122 153 defer cancel() 123 154 155 155 + var resolvedCaches []ResolvedCache 156 156 + if cacheEnabled && len(w.Caches) > 0 { 157 157 + if cachesSupported { 158 158 + resolvedCaches = ResolveCaches(ctx, l, db, pipeline.RepoDid.String(), w.Engine, cacheRepoPath, cacheRev, w.Caches) 159 159 + wfLogger.ControlWriter(CacheRestoreStepIdx, CacheRestoreStep, models.StepStatusStart).Write([]byte{0}) 160 160 + // caches are an optimization, never a reason to fail the workflow 161 161 + restoreStore := cacheStoreForRestore(cacheStore, db, l, resolvedCaches) 162 162 + if err := cacheRunner.RestoreCache(ctx, wid, &w, restoreStore, resolvedCaches, wfLogger); err != nil { 163 163 + l.Warn("cache restore failed", "wid", wid, "err", err) 164 164 + } 165 165 + wfLogger.ControlWriter(CacheRestoreStepIdx, CacheRestoreStep, models.StepStatusEnd).Write([]byte{0}) 166 166 + } else { 167 167 + l.Warn("engine does not support caches, skipping restore", "wid", wid) 168 168 + } 169 169 + } 170 170 + 171 171 + // dont save on timeouts, their context is already dead 172 172 + saveCaches := func(failed bool) { 173 173 + toSave := resolvedCaches[:0] 174 174 + for _, rc := range resolvedCaches { 175 175 + if rc.saveOn(failed) { 176 176 + toSave = append(toSave, rc) 177 177 + } 178 178 + } 179 179 + if len(toSave) == 0 { 180 180 + return 181 181 + } 182 182 + wfLogger.ControlWriter(CacheSaveStepIdx, CacheSaveStep, models.StepStatusStart).Write([]byte{0}) 183 183 + saveStore, err := prepareCacheSaves(ctx, cacheStore, db, l, cacheOwnerDID, pipeline.RepoDid.String(), w.Engine, toSave) 184 184 + if err != nil { 185 185 + l.Warn("cache metadata setup failed", "wid", wid, "err", err) 186 186 + } else { 187 187 + if err := cacheRunner.SaveCache(ctx, wid, &w, saveStore, toSave, wfLogger); err != nil { 188 188 + l.Warn("cache save failed", "wid", wid, "err", err) 189 189 + } 190 190 + saveStore.cleanup(context.WithoutCancel(ctx)) 191 191 + } 192 192 + wfLogger.ControlWriter(CacheSaveStepIdx, CacheSaveStep, models.StepStatusEnd).Write([]byte{0}) 193 193 + } 194 194 + 124 195 for stepIdx, step := range w.Steps { 125 196 // log start of step 126 197 if wfLogger != nil { ··· 145 216 l.Error("failed to set workflow status to timeout", "wid", wid, "err", dbErr) 146 217 } 147 218 } else { 219 219 + saveCaches(true) 148 220 dbErr := db.StatusFailed(wid, err.Error(), -1, n) 149 221 if dbErr != nil { 150 222 l.Error("failed to set workflow status to failed", "wid", wid, "err", dbErr) ··· 153 225 return 154 226 } 155 227 } 228 228 + 229 229 + saveCaches(false) 156 230 157 231 if finalizer, ok := eng.(workflowFinalizer); ok { 158 232 if err := finalizer.FinalizeWorkflow(ctx, wid, &w, wfLogger); err != nil {

+63

spindle/models/cache.go

Reviewed

··· 1 1 + package models 2 2 + 3 3 + import ( 4 4 + "fmt" 5 5 + "regexp" 6 6 + "strings" 7 7 + ) 8 8 + 9 9 + type CacheEntry struct { 10 10 + Key string `yaml:"key"` 11 11 + Hash []string `yaml:"hash"` 12 12 + Paths []string `yaml:"paths"` 13 13 + // 1 is fastest, 19 is smallest, 0 is the zstd default 14 14 + CompressionLevel int `yaml:"compression-level"` 15 15 + // on-success is the default, always also saves on failed runs 16 16 + When string `yaml:"when"` 17 17 + } 18 18 + 19 19 + // keys become storage paths, so no slashes 20 20 + var cacheKeyRe = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$`) 21 21 + 22 22 + func (c CacheEntry) Validate() error { 23 23 + if !cacheKeyRe.MatchString(c.Key) { 24 24 + return fmt.Errorf("cache: invalid key %q (allowed: letters, digits, '.', '_', '-')", c.Key) 25 25 + } 26 26 + for _, f := range c.Hash { 27 27 + // rev-parse needs plain repo-relative paths, not pathspecs 28 28 + if f == "" || strings.HasPrefix(f, "/") || strings.HasPrefix(f, "..") { 29 29 + return fmt.Errorf("cache %q: hash path %q is not repo-relative", c.Key, f) 30 30 + } 31 31 + if strings.ContainsAny(f, ": \t\n\"'`$\\*?[") || strings.Contains(f, "/../") || strings.HasSuffix(f, "/..") { 32 32 + return fmt.Errorf("cache %q: hash path %q contains unsupported characters", c.Key, f) 33 33 + } 34 34 + } 35 35 + if len(c.Paths) == 0 { 36 36 + return fmt.Errorf("cache %q: no paths", c.Key) 37 37 + } 38 38 + if c.CompressionLevel < 0 || c.CompressionLevel > 19 { 39 39 + return fmt.Errorf("cache %q: compression-level %d out of range (1-19)", c.Key, c.CompressionLevel) 40 40 + } 41 41 + switch c.When { 42 42 + case "", "on-success", "always": 43 43 + default: 44 44 + return fmt.Errorf("cache %q: when %q is not one of on-success, always", c.Key, c.When) 45 45 + } 46 46 + seen := make(map[string]bool, len(c.Paths)) 47 47 + for _, p := range c.Paths { 48 48 + if p == "" { 49 49 + return fmt.Errorf("cache %q: empty path", c.Key) 50 50 + } 51 51 + if strings.ContainsAny(p, " \t\n\"'`$\\") { 52 52 + return fmt.Errorf("cache %q: path %q contains unsupported characters", c.Key, p) 53 53 + } 54 54 + if strings.Contains(p, "..") { 55 55 + return fmt.Errorf("cache %q: path %q must not contain '..'", c.Key, p) 56 56 + } 57 57 + if seen[p] { 58 58 + return fmt.Errorf("cache %q: duplicate path %q", c.Key, p) 59 59 + } 60 60 + seen[p] = true 61 61 + } 62 62 + return nil 63 63 + }

+49

spindle/models/cache_test.go

Reviewed

··· 1 1 + package models 2 2 + 3 3 + import ( 4 4 + "strings" 5 5 + "testing" 6 6 + ) 7 7 + 8 8 + func TestCacheEntryValidate(t *testing.T) { 9 9 + valid := CacheEntry{Key: "go-mod-v1", Hash: []string{"go.sum", "sub/dir/package-lock.json"}, Paths: []string{"/workspace/go/pkg/mod", "/root/.cache"}} 10 10 + if err := valid.Validate(); err != nil { 11 11 + t.Fatalf("valid entry: %v", err) 12 12 + } 13 13 + 14 14 + cases := []struct { 15 15 + name string 16 16 + entry CacheEntry 17 17 + want string 18 18 + }{ 19 19 + {"empty key", CacheEntry{Key: "", Paths: []string{"/x"}}, "invalid key"}, 20 20 + {"key with slash", CacheEntry{Key: "a/b", Paths: []string{"/x"}}, "invalid key"}, 21 21 + {"no paths", CacheEntry{Key: "ok"}, "no paths"}, 22 22 + {"empty path", CacheEntry{Key: "ok", Paths: []string{""}}, "empty path"}, 23 23 + {"path with space", CacheEntry{Key: "ok", Paths: []string{"/my dir"}}, "unsupported characters"}, 24 24 + {"path with quote", CacheEntry{Key: "ok", Paths: []string{"/x'$(rm)"}}, "unsupported characters"}, 25 25 + {"path traversal", CacheEntry{Key: "ok", Paths: []string{"/x/../y"}}, ".."}, 26 26 + {"duplicate path", CacheEntry{Key: "ok", Paths: []string{"/x", "/x"}}, "duplicate"}, 27 27 + {"level too high", CacheEntry{Key: "ok", Paths: []string{"/x"}, CompressionLevel: 20}, "out of range"}, 28 28 + {"negative level", CacheEntry{Key: "ok", Paths: []string{"/x"}, CompressionLevel: -1}, "out of range"}, 29 29 + {"bad when", CacheEntry{Key: "ok", Paths: []string{"/x"}, When: "sometimes"}, "not one of"}, 30 30 + {"absolute hash path", CacheEntry{Key: "ok", Hash: []string{"/go.sum"}, Paths: []string{"/x"}}, "not repo-relative"}, 31 31 + {"empty hash path", CacheEntry{Key: "ok", Hash: []string{""}, Paths: []string{"/x"}}, "not repo-relative"}, 32 32 + {"hash path traversal", CacheEntry{Key: "ok", Hash: []string{"../secret"}, Paths: []string{"/x"}}, "not repo-relative"}, 33 33 + {"hash path inner traversal", CacheEntry{Key: "ok", Hash: []string{"a/../../b"}, Paths: []string{"/x"}}, "unsupported characters"}, 34 34 + {"hash path with colon", CacheEntry{Key: "ok", Hash: []string{"rev:go.sum"}, Paths: []string{"/x"}}, "unsupported characters"}, 35 35 + {"hash path with space", CacheEntry{Key: "ok", Hash: []string{"my lock"}, Paths: []string{"/x"}}, "unsupported characters"}, 36 36 + {"hash path with glob", CacheEntry{Key: "ok", Hash: []string{"*.sum"}, Paths: []string{"/x"}}, "unsupported characters"}, 37 37 + } 38 38 + for _, tc := range cases { 39 39 + err := tc.entry.Validate() 40 40 + if err == nil || !strings.Contains(err.Error(), tc.want) { 41 41 + t.Errorf("%s: got %v, want error containing %q", tc.name, err, tc.want) 42 42 + } 43 43 + } 44 44 + 45 45 + ok := CacheEntry{Key: "go-mod", Hash: []string{"go.sum"}, Paths: []string{"node_modules", "/root/.cache"}, CompressionLevel: 19, When: "always"} 46 46 + if err := ok.Validate(); err != nil { 47 47 + t.Errorf("relative and absolute paths should validate, got %v", err) 48 48 + } 49 49 + }

+2 -3

spindle/models/clone.go

Reviewed

··· 47 47 return CloneStep{} 48 48 } 49 49 50 50 - commitSHA, err := extractCommitSHA(tr) 50 50 + commitSHA, err := ExtractCommitSHA(tr) 51 51 if err != nil { 52 52 return CloneStep{ 53 53 kind: StepKindSystem, ··· 83 83 } 84 84 } 85 85 86 86 - // extractCommitSHA extracts the commit SHA from trigger metadata based on trigger type 87 87 - func extractCommitSHA(tr tangled.Pipeline_TriggerMetadata) (string, error) { 86 86 + func ExtractCommitSHA(tr tangled.Pipeline_TriggerMetadata) (string, error) { 88 87 switch workflow.TriggerKind(tr.Kind) { 89 88 case workflow.TriggerKindPush: 90 89 if tr.Push == nil {

+9 -1

spindle/models/pipeline.go

Reviewed

··· 1 1 package models 2 2 3 3 - import "github.com/bluesky-social/indigo/atproto/syntax" 3 3 + import ( 4 4 + "github.com/bluesky-social/indigo/atproto/syntax" 5 5 + 6 6 + "tangled.org/core/api/tangled" 7 7 + ) 4 8 5 9 type Pipeline struct { 6 10 RepoDid syntax.DID 7 11 Workflows map[Engine][]Workflow 8 12 // whether the code being ran was checked out from RepoDid itself 9 13 TrustedSource bool 14 14 + // used to resolve cache hash files against the checkout the workflow builds 15 15 + TriggerMetadata *tangled.Pipeline_TriggerMetadata 10 16 } 11 17 12 18 type Step interface { ··· 29 35 Name string 30 36 Data any 31 37 Environment map[string]string 38 38 + Caches []CacheEntry 39 39 + Engine string 32 40 }

+87

spindle/storage/disk.go

Reviewed

··· 1 1 + package storage 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "fmt" 6 6 + "io" 7 7 + "os" 8 8 + "path/filepath" 9 9 + ) 10 10 + 11 11 + type Disk struct { 12 12 + root string 13 13 + } 14 14 + 15 15 + func NewDisk(root string) (*Disk, error) { 16 16 + if root == "" { 17 17 + return nil, fmt.Errorf("storage: disk backend requires a directory") 18 18 + } 19 19 + if err := os.MkdirAll(root, 0o755); err != nil { 20 20 + return nil, fmt.Errorf("storage: create disk root: %w", err) 21 21 + } 22 22 + return &Disk{root: filepath.Clean(root)}, nil 23 23 + } 24 24 + 25 25 + func (d *Disk) path(key string) (string, error) { 26 26 + if err := ValidateKey(key); err != nil { 27 27 + return "", err 28 28 + } 29 29 + return filepath.Join(d.root, filepath.FromSlash(key)), nil 30 30 + } 31 31 + 32 32 + func (d *Disk) Get(_ context.Context, key string) (io.ReadCloser, error) { 33 33 + p, err := d.path(key) 34 34 + if err != nil { 35 35 + return nil, err 36 36 + } 37 37 + f, err := os.Open(p) 38 38 + if err != nil { 39 39 + if os.IsNotExist(err) { 40 40 + return nil, ErrNotExist 41 41 + } 42 42 + return nil, fmt.Errorf("storage: get %q: %w", key, err) 43 43 + } 44 44 + return f, nil 45 45 + } 46 46 + 47 47 + func (d *Disk) Put(_ context.Context, key string, r io.Reader) error { 48 48 + p, err := d.path(key) 49 49 + if err != nil { 50 50 + return err 51 51 + } 52 52 + if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil { 53 53 + return fmt.Errorf("storage: put %q: %w", key, err) 54 54 + } 55 55 + // dont let readers see a partial object 56 56 + tmp, err := os.CreateTemp(filepath.Dir(p), ".tmp-*") 57 57 + if err != nil { 58 58 + return fmt.Errorf("storage: put %q: %w", key, err) 59 59 + } 60 60 + tmpName := tmp.Name() 61 61 + defer os.Remove(tmpName) 62 62 + if _, err := io.Copy(tmp, r); err != nil { 63 63 + _ = tmp.Close() 64 64 + return fmt.Errorf("storage: put %q: %w", key, err) 65 65 + } 66 66 + if err := tmp.Close(); err != nil { 67 67 + return fmt.Errorf("storage: put %q: %w", key, err) 68 68 + } 69 69 + if err := os.Rename(tmpName, p); err != nil { 70 70 + return fmt.Errorf("storage: put %q: %w", key, err) 71 71 + } 72 72 + return nil 73 73 + } 74 74 + 75 75 + func (d *Disk) Delete(_ context.Context, key string) error { 76 76 + p, err := d.path(key) 77 77 + if err != nil { 78 78 + return err 79 79 + } 80 80 + if err := os.Remove(p); err != nil { 81 81 + if os.IsNotExist(err) { 82 82 + return nil 83 83 + } 84 84 + return fmt.Errorf("storage: delete %q: %w", key, err) 85 85 + } 86 86 + return nil 87 87 + }

+102

spindle/storage/s3.go

Reviewed

··· 1 1 + package storage 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "errors" 6 6 + "fmt" 7 7 + "io" 8 8 + "strings" 9 9 + 10 10 + awsconfig "github.com/aws/aws-sdk-go-v2/config" 11 11 + "github.com/aws/aws-sdk-go-v2/service/s3" 12 12 + "github.com/aws/aws-sdk-go-v2/service/s3/types" 13 13 + ) 14 14 + 15 15 + type S3 struct { 16 16 + bucket string 17 17 + prefix string 18 18 + client *s3.Client 19 19 + } 20 20 + 21 21 + func NewS3(ctx context.Context, bucket, prefix string) (*S3, error) { 22 22 + if bucket == "" { 23 23 + return nil, fmt.Errorf("storage: s3 backend requires a bucket") 24 24 + } 25 25 + sdkConfig, err := awsconfig.LoadDefaultConfig(ctx) 26 26 + if err != nil { 27 27 + return nil, fmt.Errorf("storage: load s3 config: %w", err) 28 28 + } 29 29 + client := s3.NewFromConfig(sdkConfig) 30 30 + versioning, err := client.GetBucketVersioning(ctx, &s3.GetBucketVersioningInput{ 31 31 + Bucket: &bucket, 32 32 + }) 33 33 + if err != nil { 34 34 + return nil, fmt.Errorf("storage: check s3 bucket versioning: %w", err) 35 35 + } 36 36 + if versioning.Status != "" { 37 37 + return nil, fmt.Errorf("storage: s3 cache bucket must not use versioning") 38 38 + } 39 39 + return &S3{ 40 40 + bucket: bucket, 41 41 + prefix: strings.Trim(prefix, "/"), 42 42 + client: client, 43 43 + }, nil 44 44 + } 45 45 + 46 46 + func (s *S3) fullKey(key string) (string, error) { 47 47 + if err := ValidateKey(key); err != nil { 48 48 + return "", err 49 49 + } 50 50 + if s.prefix == "" { 51 51 + return key, nil 52 52 + } 53 53 + return s.prefix + "/" + key, nil 54 54 + } 55 55 + 56 56 + func (s *S3) Get(ctx context.Context, key string) (io.ReadCloser, error) { 57 57 + full, err := s.fullKey(key) 58 58 + if err != nil { 59 59 + return nil, err 60 60 + } 61 61 + res, err := s.client.GetObject(ctx, &s3.GetObjectInput{ 62 62 + Bucket: &s.bucket, 63 63 + Key: &full, 64 64 + }) 65 65 + if err != nil { 66 66 + var nsk *types.NoSuchKey 67 67 + if errors.As(err, &nsk) { 68 68 + return nil, ErrNotExist 69 69 + } 70 70 + return nil, fmt.Errorf("storage: get %q: %w", key, err) 71 71 + } 72 72 + return res.Body, nil 73 73 + } 74 74 + 75 75 + func (s *S3) Put(ctx context.Context, key string, r io.Reader) error { 76 76 + full, err := s.fullKey(key) 77 77 + if err != nil { 78 78 + return err 79 79 + } 80 80 + if _, err := s.client.PutObject(ctx, &s3.PutObjectInput{ 81 81 + Bucket: &s.bucket, 82 82 + Key: &full, 83 83 + Body: r, 84 84 + }); err != nil { 85 85 + return fmt.Errorf("storage: put %q: %w", key, err) 86 86 + } 87 87 + return nil 88 88 + } 89 89 + 90 90 + func (s *S3) Delete(ctx context.Context, key string) error { 91 91 + full, err := s.fullKey(key) 92 92 + if err != nil { 93 93 + return err 94 94 + } 95 95 + if _, err := s.client.DeleteObject(ctx, &s3.DeleteObjectInput{ 96 96 + Bucket: &s.bucket, 97 97 + Key: &full, 98 98 + }); err != nil { 99 99 + return fmt.Errorf("storage: delete %q: %w", key, err) 100 100 + } 101 101 + return nil 102 102 + }

+52

spindle/storage/storage.go

Reviewed

··· 1 1 + package storage 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "errors" 6 6 + "fmt" 7 7 + "io" 8 8 + "path/filepath" 9 9 + "regexp" 10 10 + "strings" 11 11 + 12 12 + "tangled.org/core/spindle/config" 13 13 + ) 14 14 + 15 15 + var ErrNotExist = errors.New("storage: object does not exist") 16 16 + 17 17 + type Storage interface { 18 18 + Get(ctx context.Context, key string) (io.ReadCloser, error) 19 19 + Put(ctx context.Context, key string, r io.Reader) error 20 20 + Delete(ctx context.Context, key string) error 21 21 + } 22 22 + 23 23 + var keyRe = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:/@%+=-]{0,511}$`) 24 24 + 25 25 + func ValidateKey(key string) error { 26 26 + if !keyRe.MatchString(key) { 27 27 + return fmt.Errorf("storage: invalid key %q", key) 28 28 + } 29 29 + for _, seg := range strings.Split(key, "/") { 30 30 + if seg == "" || seg == "." || seg == ".." { 31 31 + return fmt.Errorf("storage: invalid key %q", key) 32 32 + } 33 33 + } 34 34 + return nil 35 35 + } 36 36 + 37 37 + func New(ctx context.Context, cfg *config.Config) (Storage, error) { 38 38 + switch cfg.Cache.Backend { 39 39 + case "": 40 40 + return nil, nil 41 41 + case "disk": 42 42 + dir := cfg.Cache.DiskDir 43 43 + if dir == "" { 44 44 + dir = filepath.Join(filepath.Dir(cfg.Server.DBPath), "cache") 45 45 + } 46 46 + return NewDisk(dir) 47 47 + case "s3": 48 48 + return NewS3(ctx, cfg.Cache.S3Bucket, cfg.Cache.S3Prefix) 49 49 + default: 50 50 + return nil, fmt.Errorf("storage: unknown backend %q", cfg.Cache.Backend) 51 51 + } 52 52 + }

+101

spindle/storage/storage_test.go

Reviewed

··· 1 1 + package storage 2 2 + 3 3 + import ( 4 4 + "context" 5 5 + "errors" 6 6 + "io" 7 7 + "strings" 8 8 + "testing" 9 9 + ) 10 10 + 11 11 + func TestValidateKey(t *testing.T) { 12 12 + valid := []string{ 13 13 + "did:plc:xyz123/go-mod-v1", 14 14 + "did:web:spindle.example.com/cache.tar", 15 15 + "abc", 16 16 + "a/b/c/d", 17 17 + } 18 18 + for _, k := range valid { 19 19 + if err := ValidateKey(k); err != nil { 20 20 + t.Errorf("ValidateKey(%q) = %v, want nil", k, err) 21 21 + } 22 22 + } 23 23 + 24 24 + invalid := []string{ 25 25 + "", 26 26 + "../escape", 27 27 + "a/../../b", 28 28 + "/leading", 29 29 + "trailing/", 30 30 + "double//slash", 31 31 + "with space", 32 32 + "with\\backslash", 33 33 + "-leading-dash", 34 34 + } 35 35 + for _, k := range invalid { 36 36 + if err := ValidateKey(k); err == nil { 37 37 + t.Errorf("ValidateKey(%q) = nil, want error", k) 38 38 + } 39 39 + } 40 40 + } 41 41 + 42 42 + func TestDiskRoundTrip(t *testing.T) { 43 43 + ctx := context.Background() 44 44 + d, err := NewDisk(t.TempDir()) 45 45 + if err != nil { 46 46 + t.Fatal(err) 47 47 + } 48 48 + 49 49 + key := "did:plc:xyz/go-mod-v1" 50 50 + if err := d.Put(ctx, key, strings.NewReader("archive-bytes")); err != nil { 51 51 + t.Fatal(err) 52 52 + } 53 53 + 54 54 + rc, err := d.Get(ctx, key) 55 55 + if err != nil { 56 56 + t.Fatal(err) 57 57 + } 58 58 + got, err := io.ReadAll(rc) 59 59 + rc.Close() 60 60 + if err != nil { 61 61 + t.Fatal(err) 62 62 + } 63 63 + if string(got) != "archive-bytes" { 64 64 + t.Fatalf("got %q, want %q", got, "archive-bytes") 65 65 + } 66 66 + 67 67 + // overwrite 68 68 + if err := d.Put(ctx, key, strings.NewReader("new-bytes")); err != nil { 69 69 + t.Fatal(err) 70 70 + } 71 71 + rc, _ = d.Get(ctx, key) 72 72 + got, _ = io.ReadAll(rc) 73 73 + rc.Close() 74 74 + if string(got) != "new-bytes" { 75 75 + t.Fatalf("overwrite: got %q, want %q", got, "new-bytes") 76 76 + } 77 77 + 78 78 + if err := d.Delete(ctx, key); err != nil { 79 79 + t.Fatalf("delete: %v", err) 80 80 + } 81 81 + if _, err := d.Get(ctx, key); !errors.Is(err, ErrNotExist) { 82 82 + t.Fatalf("get deleted: got %v, want ErrNotExist", err) 83 83 + } 84 84 + if err := d.Delete(ctx, key); err != nil { 85 85 + t.Fatalf("delete missing: %v", err) 86 86 + } 87 87 + } 88 88 + 89 89 + func TestDiskRejectsTraversal(t *testing.T) { 90 90 + ctx := context.Background() 91 91 + d, err := NewDisk(t.TempDir()) 92 92 + if err != nil { 93 93 + t.Fatal(err) 94 94 + } 95 95 + if err := d.Put(ctx, "../evil", strings.NewReader("x")); err == nil { 96 96 + t.Fatal("put with traversal key succeeded") 97 97 + } 98 98 + if _, err := d.Get(ctx, "../evil"); err == nil { 99 99 + t.Fatal("get with traversal key succeeded") 100 100 + } 101 101 + }

+125

spindle/engines/microvm/cache.go

Reviewed

··· 1 1 + package microvm 2 2 + 3 3 + import ( 4 4 + "bufio" 5 5 + "context" 6 6 + "fmt" 7 7 + "io" 8 8 + 9 9 + agentv1 "tangled.org/core/spindle/agentproto/gen" 10 10 + "tangled.org/core/spindle/engine" 11 11 + "tangled.org/core/spindle/models" 12 12 + "tangled.org/core/spindle/storage" 13 13 + ) 14 14 + 15 15 + func (e *Engine) RestoreCache(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, store storage.Storage, caches []engine.ResolvedCache, wfLogger models.WorkflowLogger) error { 16 16 + state, ok := wf.Data.(*workflowState) 17 17 + if !ok || state == nil || state.Agent == nil { 18 18 + return fmt.Errorf("microVM workflow is not connected to agent") 19 19 + } 20 20 + 21 21 + out := wfLogger.DataWriter(engine.CacheRestoreStepIdx, "stdout") 22 22 + for _, entry := range caches { 23 23 + if err := ctx.Err(); err != nil { 24 24 + return err 25 25 + } 26 26 + if entry.RestoreKey == "" { 27 27 + fmt.Fprintf(out, "cache %q: miss\n", entry.Key) 28 28 + continue 29 29 + } 30 30 + if entry.RestoreName != "" { 31 31 + fmt.Fprintf(out, "cache %q: restoring from %q\n", entry.Key, entry.RestoreName) 32 32 + } 33 33 + 34 34 + rc, err := store.Get(ctx, entry.RestoreKey) 35 35 + if err != nil { 36 36 + fmt.Fprintf(out, "cache %q: fetch failed: %v\n", entry.Key, err) 37 37 + continue 38 38 + } 39 39 + br := bufio.NewReader(rc) 40 40 + decompress := engine.CacheDecompressCmd(br) 41 41 + 42 42 + var restored int64 43 43 + exit, err := state.Agent.Exec(ctx, AgentExec{ 44 44 + ID: fmt.Sprintf("%s-cache-restore", wid.String()), 45 45 + ExecStart: cacheExecStart(state, fmt.Sprintf("set -o pipefail\n%s | tar -x -C /", decompress)), 46 46 + Stdin: &countingReader{r: br, n: &restored}, 47 47 + Stderr: out, 48 48 + }) 49 49 + rc.Close() 50 50 + if err != nil { 51 51 + fmt.Fprintf(out, "cache %q: restore failed: %v\n", entry.Key, err) 52 52 + continue 53 53 + } 54 54 + if exit != 0 { 55 55 + fmt.Fprintf(out, "cache %q: restore failed: guest exited %d\n", entry.Key, exit) 56 56 + continue 57 57 + } 58 58 + fmt.Fprintf(out, "cache %q: restored %d bytes\n", entry.Key, restored) 59 59 + } 60 60 + return nil 61 61 + } 62 62 + 63 63 + func (e *Engine) SaveCache(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, store storage.Storage, caches []engine.ResolvedCache, wfLogger models.WorkflowLogger) error { 64 64 + state, ok := wf.Data.(*workflowState) 65 65 + if !ok || state == nil || state.Agent == nil { 66 66 + return fmt.Errorf("microVM workflow is not connected to agent") 67 67 + } 68 68 + 69 69 + out := wfLogger.DataWriter(engine.CacheSaveStepIdx, "stdout") 70 70 + for _, entry := range caches { 71 71 + if err := ctx.Err(); err != nil { 72 72 + return err 73 73 + } 74 74 + 75 75 + script := engine.CacheSaveScript(entry.Paths, guestWorkDir, entry.CompressionLevel) 76 76 + 77 77 + up := engine.NewCacheUpload(ctx, store, entry.SaveKey) 78 78 + exit, execErr := state.Agent.Exec(ctx, AgentExec{ 79 79 + ID: fmt.Sprintf("%s-cache-save", wid.String()), 80 80 + ExecStart: cacheExecStart(state, script), 81 81 + Stdout: up.Writer, 82 82 + Stderr: out, 83 83 + }) 84 84 + switch { 85 85 + case exit == engine.CacheExitNoPaths: 86 86 + up.Abort(fmt.Errorf("guest exited %d", exit)) 87 87 + fmt.Fprintf(out, "cache %q: nothing to save\n", entry.Key) 88 88 + continue 89 89 + case exit == engine.CacheExitNoCompressor: 90 90 + up.Abort(fmt.Errorf("guest exited %d", exit)) 91 91 + fmt.Fprintf(out, "cache %q: zstd not available in image; skipping\n", entry.Key) 92 92 + continue 93 93 + case execErr != nil: 94 94 + up.Abort(execErr) 95 95 + return fmt.Errorf("save cache %q: %w", entry.Key, execErr) 96 96 + case exit != 0: 97 97 + up.Abort(fmt.Errorf("guest exited %d", exit)) 98 98 + return fmt.Errorf("save cache %q: save script exited %d", entry.Key, exit) 99 99 + } 100 100 + if err := up.Finish(); err != nil { 101 101 + return fmt.Errorf("save cache %q: %w", entry.Key, err) 102 102 + } 103 103 + fmt.Fprintf(out, "cache %q: saved\n", entry.Key) 104 104 + } 105 105 + return nil 106 106 + } 107 107 + 108 108 + func cacheExecStart(state *workflowState, script string) *agentv1.ExecStart { 109 109 + return &agentv1.ExecStart{ 110 110 + Argv: []string{state.ImageSpec.Shell, "-c", script}, 111 111 + Env: guestBaseEnv(), 112 112 + User: guestWorkflowUser, 113 113 + } 114 114 + } 115 115 + 116 116 + type countingReader struct { 117 117 + r io.Reader 118 118 + n *int64 119 119 + } 120 120 + 121 121 + func (c *countingReader) Read(p []byte) (int, error) { 122 122 + n, err := c.r.Read(p) 123 123 + *c.n += int64(n) 124 124 + return n, err 125 125 + }

+18 -6

spindle/engines/microvm/engine.go

Reviewed

··· 42 42 43 43 type cleanupFunc func(context.Context) error 44 44 45 45 + // return a fresh slice since callers append to it 46 46 + func guestBaseEnv() []string { 47 47 + return []string{ 48 48 + "HOME=/workspace", 49 49 + "LOGNAME=" + guestWorkflowUser, 50 50 + "PATH=" + guestBasePATH, 51 51 + "USER=" + guestWorkflowUser, 52 52 + } 53 53 + } 54 54 + 45 55 type Engine struct { 46 56 l *slog.Logger 47 57 cfg *config.Config ··· 134 144 } 135 145 swf.Name = twf.Name 136 146 swf.Environment = dwf.Environment 147 147 + 148 148 + for _, entry := range dwf.Cache { 149 149 + if err := entry.Validate(); err != nil { 150 150 + return nil, err 151 151 + } 152 152 + } 153 153 + swf.Caches = dwf.Cache 137 154 138 155 if tpl.TriggerMetadata != nil { 139 156 if clone := models.BuildCloneStep(twf, *tpl.TriggerMetadata, e.cfg.Server.Dev); clone.Command() != "" { ··· 360 377 err := e.activateConfig(execCtx, wid, state, s, wfLogger.DataWriter(idx, "stdout")) 361 378 return e.classifyStepError(ctx, wid, step, state, stderr, vmExited, "Failed to activate config", err) 362 379 } 363 363 - env := []string{ 364 364 - "HOME=/workspace", 365 365 - "LOGNAME=" + guestWorkflowUser, 366 366 - "PATH=" + guestBasePATH, 367 367 - "USER=" + guestWorkflowUser, 368 368 - } 380 380 + env := guestBaseEnv() 369 381 for k, v := range w.Environment { 370 382 env = append(env, k+"="+v) 371 383 }

+10 -7

spindle/engines/microvm/models.go

Reviewed

··· 3 3 import ( 4 4 "fmt" 5 5 "slices" 6 6 + 7 7 + "tangled.org/core/spindle/models" 6 8 ) 7 9 8 10 type manifestWorkflow struct { 9 9 - Image string `yaml:"image"` 10 10 - Services map[string]any `yaml:"services"` 11 11 - Virtualisation map[string]any `yaml:"virtualisation"` 12 12 - Dependencies []string `yaml:"dependencies"` 13 13 - Registry map[string]any `yaml:"registry"` 14 14 - Environment map[string]string `yaml:"environment"` 15 15 - Caches map[string]string `yaml:"caches"` 11 11 + Image string `yaml:"image"` 12 12 + Services map[string]any `yaml:"services"` 13 13 + Virtualisation map[string]any `yaml:"virtualisation"` 14 14 + Dependencies []string `yaml:"dependencies"` 15 15 + Registry map[string]any `yaml:"registry"` 16 16 + Environment map[string]string `yaml:"environment"` 17 17 + Caches map[string]string `yaml:"caches"` 18 18 + Cache []models.CacheEntry `yaml:"cache"` 16 19 Steps []struct { 17 20 Name string `yaml:"name"` 18 21 Command string `yaml:"command"`

+167

spindle/engines/nixery/cache.go

Reviewed

··· 1 1 + package nixery 2 2 + 3 3 + import ( 4 4 + "bufio" 5 5 + "context" 6 6 + "fmt" 7 7 + "io" 8 8 + 9 9 + "github.com/docker/docker/api/types" 10 10 + "github.com/docker/docker/api/types/container" 11 11 + "github.com/docker/docker/pkg/stdcopy" 12 12 + 13 13 + "tangled.org/core/spindle/engine" 14 14 + "tangled.org/core/spindle/models" 15 15 + "tangled.org/core/spindle/storage" 16 16 + ) 17 17 + 18 18 + func (e *Engine) baseEnv() EnvVars { 19 19 + envs := EnvVars{} 20 20 + envs.AddEnv("HOME", homeDir) 21 21 + envs.AddEnv("PATH", fmt.Sprintf("%s/.nix-profile/bin:/nix/var/nix/profiles/default/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin", homeDir)) 22 22 + return envs 23 23 + } 24 24 + 25 25 + func (e *Engine) execAttached(ctx context.Context, containerID string, opts container.ExecOptions) (string, types.HijackedResponse, error) { 26 26 + execResp, err := e.docker.ContainerExecCreate(ctx, containerID, opts) 27 27 + if err != nil { 28 28 + return "", types.HijackedResponse{}, fmt.Errorf("create exec: %w", err) 29 29 + } 30 30 + attach, err := e.docker.ContainerExecAttach(ctx, execResp.ID, container.ExecAttachOptions{}) 31 31 + if err != nil { 32 32 + return "", types.HijackedResponse{}, fmt.Errorf("attach exec: %w", err) 33 33 + } 34 34 + return execResp.ID, attach, nil 35 35 + } 36 36 + 37 37 + func (e *Engine) containerID(wf *models.Workflow) (string, error) { 38 38 + addl, ok := wf.Data.(addlFields) 39 39 + if !ok || addl.container == "" { 40 40 + return "", fmt.Errorf("nixery workflow has no container") 41 41 + } 42 42 + return addl.container, nil 43 43 + } 44 44 + 45 45 + func (e *Engine) RestoreCache(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, store storage.Storage, caches []engine.ResolvedCache, wfLogger models.WorkflowLogger) error { 46 46 + containerID, err := e.containerID(wf) 47 47 + if err != nil { 48 48 + return err 49 49 + } 50 50 + 51 51 + out := wfLogger.DataWriter(engine.CacheRestoreStepIdx, "stdout") 52 52 + for _, entry := range caches { 53 53 + if err := ctx.Err(); err != nil { 54 54 + return err 55 55 + } 56 56 + if entry.RestoreKey == "" { 57 57 + fmt.Fprintf(out, "cache %q: miss\n", entry.Key) 58 58 + continue 59 59 + } 60 60 + if entry.RestoreName != "" { 61 61 + fmt.Fprintf(out, "cache %q: restoring from %q\n", entry.Key, entry.RestoreName) 62 62 + } 63 63 + 64 64 + rc, err := store.Get(ctx, entry.RestoreKey) 65 65 + if err != nil { 66 66 + fmt.Fprintf(out, "cache %q: fetch failed: %v\n", entry.Key, err) 67 67 + continue 68 68 + } 69 69 + 70 70 + br := bufio.NewReader(rc) 71 71 + execID, attach, err := e.execAttached(ctx, containerID, container.ExecOptions{ 72 72 + Cmd: []string{"bash", "-c", engine.CacheDecompressCmd(br) + " | tar -x -C /"}, 73 73 + Env: e.baseEnv(), 74 74 + AttachStdin: true, 75 75 + AttachStdout: true, 76 76 + AttachStderr: true, 77 77 + }) 78 78 + if err != nil { 79 79 + rc.Close() 80 80 + return fmt.Errorf("restore cache %q: %w", entry.Key, err) 81 81 + } 82 82 + 83 83 + // drain this now or tar can block on stderr before reading stdin 84 84 + copyDone := make(chan error, 1) 85 85 + go func() { 86 86 + _, err := io.Copy(attach.Conn, br) 87 87 + _ = attach.CloseWrite() 88 88 + copyDone <- err 89 89 + }() 90 90 + _, _ = stdcopy.StdCopy(out, out, attach.Reader) 91 91 + copyErr := <-copyDone 92 92 + rc.Close() 93 93 + attach.Close() 94 94 + if copyErr != nil { 95 95 + return fmt.Errorf("restore cache %q: stream archive: %w", entry.Key, copyErr) 96 96 + } 97 97 + 98 98 + inspect, err := e.docker.ContainerExecInspect(ctx, execID) 99 99 + if err != nil { 100 100 + return fmt.Errorf("restore cache %q: %w", entry.Key, err) 101 101 + } 102 102 + if inspect.ExitCode != 0 { 103 103 + fmt.Fprintf(out, "cache %q: extract failed (exit %d)\n", entry.Key, inspect.ExitCode) 104 104 + continue 105 105 + } 106 106 + fmt.Fprintf(out, "cache %q: restored\n", entry.Key) 107 107 + } 108 108 + return nil 109 109 + } 110 110 + 111 111 + func (e *Engine) SaveCache(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, store storage.Storage, caches []engine.ResolvedCache, wfLogger models.WorkflowLogger) error { 112 112 + containerID, err := e.containerID(wf) 113 113 + if err != nil { 114 114 + return err 115 115 + } 116 116 + 117 117 + out := wfLogger.DataWriter(engine.CacheSaveStepIdx, "stdout") 118 118 + for _, entry := range caches { 119 119 + if err := ctx.Err(); err != nil { 120 120 + return err 121 121 + } 122 122 + 123 123 + script := engine.CacheSaveScript(entry.Paths, workspaceDir, entry.CompressionLevel) 124 124 + 125 125 + execID, attach, err := e.execAttached(ctx, containerID, container.ExecOptions{ 126 126 + Cmd: []string{"bash", "-c", script}, 127 127 + Env: e.baseEnv(), 128 128 + AttachStdout: true, 129 129 + AttachStderr: true, 130 130 + }) 131 131 + if err != nil { 132 132 + return fmt.Errorf("save cache %q: %w", entry.Key, err) 133 133 + } 134 134 + 135 135 + up := engine.NewCacheUpload(ctx, store, entry.SaveKey) 136 136 + 137 137 + // StdCopy only returns once the archive is fully written 138 138 + _, copyErr := stdcopy.StdCopy(up.Writer, out, attach.Reader) 139 139 + attach.Close() 140 140 + inspect, inspectErr := e.docker.ContainerExecInspect(ctx, execID) 141 141 + 142 142 + switch { 143 143 + case inspectErr == nil && inspect.ExitCode == engine.CacheExitNoPaths: 144 144 + up.Abort(fmt.Errorf("no cache paths")) 145 145 + fmt.Fprintf(out, "cache %q: nothing to save\n", entry.Key) 146 146 + continue 147 147 + case inspectErr == nil && inspect.ExitCode == engine.CacheExitNoCompressor: 148 148 + up.Abort(fmt.Errorf("zstd not available")) 149 149 + fmt.Fprintf(out, "cache %q: zstd not available in image; skipping\n", entry.Key) 150 150 + continue 151 151 + case copyErr != nil: 152 152 + up.Abort(copyErr) 153 153 + return fmt.Errorf("save cache %q: stream archive: %w", entry.Key, copyErr) 154 154 + case inspectErr != nil: 155 155 + up.Abort(inspectErr) 156 156 + return fmt.Errorf("save cache %q: %w", entry.Key, inspectErr) 157 157 + case inspect.ExitCode != 0: 158 158 + up.Abort(fmt.Errorf("exited %d", inspect.ExitCode)) 159 159 + return fmt.Errorf("save cache %q: tar exited %d", entry.Key, inspect.ExitCode) 160 160 + } 161 161 + if err := up.Finish(); err != nil { 162 162 + return fmt.Errorf("save cache %q: %w", entry.Key, err) 163 163 + } 164 164 + fmt.Fprintf(out, "cache %q: saved\n", entry.Key) 165 165 + } 166 166 + return nil 167 167 + }

+9 -4

spindle/engines/nixery/engine.go

Reviewed

··· 91 91 } `yaml:"steps"` 92 92 Dependencies map[string][]string `yaml:"dependencies"` 93 93 Environment map[string]string `yaml:"environment"` 94 94 + Cache []models.CacheEntry `yaml:"cache"` 94 95 }{} 95 96 if err := engine.DescribeManifestError(twf.Raw, dwf); err != nil { 96 97 return nil, err ··· 109 110 } 110 111 swf.Name = twf.Name 111 112 swf.Environment = dwf.Environment 113 113 + for _, entry := range dwf.Cache { 114 114 + if err := entry.Validate(); err != nil { 115 115 + return nil, err 116 116 + } 117 117 + } 118 118 + swf.Caches = dwf.Cache 112 119 addl.image = workflowImage(dwf.Dependencies, e.cfg.NixeryPipelines.Nixery) 113 120 114 121 if sock := e.cfg.Server.DockerSocket; sock != "" { ··· 155 162 } 156 163 157 164 // load defaults from somewhere else 158 158 - dependencies = path.Join(dependencies, "bash", "git", "coreutils", "nix") 165 165 + dependencies = path.Join(dependencies, "bash", "git", "coreutils", "gnutar", "zstd", "nix") 159 166 160 167 if runtime.GOARCH == "arm64" { 161 168 dependencies = path.Join("arm64", dependencies) ··· 409 416 } 410 417 } 411 418 412 412 - envs.AddEnv("HOME", homeDir) 413 413 - existingPath := "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" 414 414 - envs.AddEnv("PATH", fmt.Sprintf("%s/.nix-profile/bin:/nix/var/nix/profiles/default/bin:%s", homeDir, existingPath)) 419 419 + envs = append(envs, e.baseEnv()...) 415 420 if sock := e.cfg.Server.DockerSocket; sock != "" { 416 421 envs.AddEnv("DOCKER_HOST", fmt.Sprintf("unix://%s", sock)) 417 422 }