alice.pds.demo.boltless.dev
This repository has no description
spindle: add generic CI cache
Signed-off-by: dawn <dawn@tangled.org>
author
dawn
date
(Jul 25, 2026, 2:00 PM +0300)
commit
41d64257
41d64257b9348be0325c7e79ea312336c3bd78fe
parent
679087ca
679087ca63c87a0022d5faeb7afc633e31944f92
change-id
xsylqqrt
xsylqqrtmmtnoyxnukslqrqrxsvswvzs
+2493
-32
+1
docker-compose.yml
+1
docker-compose.yml
···
183
183
SPINDLE_MICROVM_PIPELINES_AGENT_PORT: "11240"
184
184
SPINDLE_S3_LOG_BUCKET: ""
185
185
SPINDLE_MICROVM_PIPELINES_ENABLE_CGROUPS: "false"
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
+76
docs/DOCS.md
···
965
965
- `TANGLED_PR_SOURCE_SHA` - The commit SHA of the source
966
966
branch
967
967
968
+
### Cache
969
+
970
+
The `cache` field lets a workflow persist directories across
971
+
pipeline runs. Before the first step, the engine looks up
972
+
each entry's key and extracts the matching archive into the
973
+
workspace; after all steps succeed, the paths are archived
974
+
again and stored back under the key.
975
+
976
+
- `key`: name this cache is saved under. Keys are scoped to
977
+
the repository and the engine specified.
978
+
- `hash`: **optional** list of repo files (lockfiles,
979
+
manifests) whose content is folded into the key. The entry
980
+
is stored as `<key>-<hash of those files>`, so editing
981
+
`go.sum` automatically rotates the cache without bumping
982
+
the key by hand. Paths are relative to the repository
983
+
root and are read from git at the commit being built.
984
+
- `paths`: paths to archive. Relative paths are anchored at
985
+
the repository checkout, the directory steps start in
986
+
(`/workspace/repo` on microvm, `/tangled/workspace` on
987
+
nixery). Absolute paths work too, for caching directories
988
+
outside the checkout, but note they name engine-specific
989
+
locations. All paths must be writable by the CI user and
990
+
contain no spaces.
991
+
- `compression-level`: **optional** zstd level, `1`
992
+
(fastest) to `19` (smallest). defaults to (`5`).
993
+
- `when`: **optional** save policy: `on-success` (the
994
+
default) or `always`.
995
+
996
+
When the exact key (or generation, with `hash`) misses, the
997
+
newest older generation under the same key is restored.
998
+
999
+
```yaml
1000
+
cache:
1001
+
- key: go-mod
1002
+
hash:
1003
+
- go.sum
1004
+
- go.mod
1005
+
paths:
1006
+
- .gocache
1007
+
```
1008
+
1009
+
Caches are only restored and saved for trusted pipelines
1010
+
(pushes and same-repository pull requests). Pipelines
1011
+
building untrusted code, like pull requests from forks,
1012
+
skip the cache entirely. Saving follows each entry's `when`
1013
+
policy, except on timeout, when nothing is saved. A cache
1014
+
miss or failure never fails the workflow.
1015
+
1016
+
The spindle operator chooses the storage backend. See
1017
+
[Running spindle](#running-spindle). If no backend is
1018
+
configured, `cache` entries are ignored.
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
+
1565
+
The generic CI cache (the workflow-level
1566
+
[`cache`](#cache) field) is configured via prefix
1567
+
`SPINDLE_CACHE_`.
1568
+
1569
+
- `SPINDLE_CACHE_BACKEND`: Storage backend, `disk` or `s3`
1570
+
(default: `""`, caching disabled).
1571
+
- `SPINDLE_CACHE_DISK_DIR`: Directory for the `disk` backend
1572
+
(default: a `cache` directory next to the spindle
1573
+
database).
1574
+
- `SPINDLE_CACHE_S3_BUCKET`: Unversioned bucket for the `s3`
1575
+
backend. Credentials come from the standard AWS chain and
1576
+
need `s3:GetBucketVersioning` in addition to object access.
1577
+
- `SPINDLE_CACHE_S3_PREFIX`: Key prefix inside the bucket
1578
+
(default: `"spindle/cache"`).
1579
+
- `SPINDLE_CACHE_RETENTION`: Time since the last restore or
1580
+
save before an entry is deleted (default: `720h`, or 30
1581
+
days). Set to `0` to keep entries indefinitely.
1582
+
- `SPINDLE_CACHE_PRUNE_INTERVAL`: How often expired entries
1583
+
are deleted (default: `1h`).
1584
+
1585
+
Cache metadata and usage are tracked in spindle's SQLite
1586
+
database. Storage backends contain opaque objects and are
1587
+
never listed during lookup or cleanup.
1512
1588
1513
1589
### Running spindle
1514
1590
+1
nix/microvm/base.nix
+1
nix/microvm/base.nix
+2
-1
nix/pkgs/spindle-alpine-image.nix
+2
-1
nix/pkgs/spindle-alpine-image.nix
···
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
-
guestTools = [nix bash git curl jq];
33
+
# zstd is not a busybox applet, and the spindle cache saves tar|zstd
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
+10
spindle/config/config.go
···
61
61
LogBucket string `env:"LOG_BUCKET"`
62
62
}
63
63
64
+
type Cache struct {
65
+
Backend string `env:"BACKEND"` // "disk" or "s3"
66
+
DiskDir string `env:"DISK_DIR"`
67
+
S3Bucket string `env:"S3_BUCKET"`
68
+
S3Prefix string `env:"S3_PREFIX, default=spindle/cache"`
69
+
Retention time.Duration `env:"RETENTION, default=720h"`
70
+
PruneInterval time.Duration `env:"PRUNE_INTERVAL, default=1h"`
71
+
}
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
+
Cache Cache `env:",prefix=SPINDLE_CACHE_"`
102
112
}
103
113
104
114
func Load(ctx context.Context) (*Config, error) {
+233
spindle/db/cache.go
+233
spindle/db/cache.go
···
1
+
package db
2
+
3
+
import (
4
+
"context"
5
+
"sort"
6
+
"time"
7
+
)
8
+
9
+
type CacheEntry struct {
10
+
ID string
11
+
StorageKey string
12
+
OwnerDID string
13
+
RepoDID string
14
+
Engine string
15
+
CacheKey string
16
+
CacheHash string
17
+
SizeBytes int64
18
+
State string
19
+
CreatedAt time.Time
20
+
LastUsedAt time.Time
21
+
}
22
+
23
+
const cacheEntryColumns = `
24
+
id, storage_key, owner_did, repo_did, engine, cache_key, cache_hash,
25
+
size_bytes, state, created_at, last_used_at`
26
+
27
+
func (d *DB) InsertCacheEntry(ctx context.Context, entry CacheEntry) error {
28
+
_, err := d.ExecContext(ctx, `
29
+
insert into cache_entries (
30
+
id, storage_key, owner_did, repo_did, engine, cache_key, cache_hash,
31
+
size_bytes, state, created_at, last_used_at
32
+
) values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`,
33
+
entry.ID,
34
+
entry.StorageKey,
35
+
entry.OwnerDID,
36
+
entry.RepoDID,
37
+
entry.Engine,
38
+
entry.CacheKey,
39
+
entry.CacheHash,
40
+
entry.SizeBytes,
41
+
entry.State,
42
+
entry.CreatedAt.UnixNano(),
43
+
entry.LastUsedAt.UnixNano(),
44
+
)
45
+
return err
46
+
}
47
+
48
+
func (d *DB) MarkCacheEntryReady(ctx context.Context, id string, sizeBytes int64, now time.Time) ([]CacheEntry, error) {
49
+
tx, err := d.BeginTx(ctx, nil)
50
+
if err != nil {
51
+
return nil, err
52
+
}
53
+
defer tx.Rollback()
54
+
55
+
var repoDID, engine, key, hash string
56
+
if err := tx.QueryRowContext(ctx, `
57
+
update cache_entries
58
+
set state = 'ready', size_bytes = ?, last_used_at = ?
59
+
where id = ? and state = 'pending'
60
+
returning repo_did, engine, cache_key, cache_hash`,
61
+
sizeBytes, now.UnixNano(), id).Scan(&repoDID, &engine, &key, &hash); err != nil {
62
+
return nil, err
63
+
}
64
+
65
+
rows, err := tx.QueryContext(ctx, `
66
+
update cache_entries
67
+
set state = 'deleting'
68
+
where repo_did = ? and engine = ? and cache_key = ? and cache_hash = ?
69
+
and state = 'ready' and id <> ?
70
+
returning `+cacheEntryColumns, repoDID, engine, key, hash, id)
71
+
if err != nil {
72
+
return nil, err
73
+
}
74
+
var superseded []CacheEntry
75
+
for rows.Next() {
76
+
entry, err := scanCacheEntry(rows)
77
+
if err != nil {
78
+
rows.Close()
79
+
return nil, err
80
+
}
81
+
superseded = append(superseded, *entry)
82
+
}
83
+
if err := rows.Close(); err != nil {
84
+
return nil, err
85
+
}
86
+
if err := rows.Err(); err != nil {
87
+
return nil, err
88
+
}
89
+
sort.Slice(superseded, func(i, j int) bool {
90
+
return superseded[i].CreatedAt.After(superseded[j].CreatedAt)
91
+
})
92
+
if err := tx.Commit(); err != nil {
93
+
return nil, err
94
+
}
95
+
return superseded, nil
96
+
}
97
+
98
+
func (d *DB) FindCacheEntry(ctx context.Context, repoDID, engine, key, hash string) (*CacheEntry, error) {
99
+
return scanCacheEntry(d.QueryRowContext(ctx, `
100
+
select `+cacheEntryColumns+`
101
+
from cache_entries
102
+
where repo_did = ? and engine = ? and cache_key = ? and cache_hash = ? and state = 'ready'
103
+
order by created_at desc
104
+
limit 1`, repoDID, engine, key, hash))
105
+
}
106
+
107
+
func (d *DB) FindFallbackCacheEntry(ctx context.Context, repoDID, engine, key, excludeHash string) (*CacheEntry, error) {
108
+
return scanCacheEntry(d.QueryRowContext(ctx, `
109
+
select `+cacheEntryColumns+`
110
+
from cache_entries
111
+
where repo_did = ? and engine = ? and cache_key = ?
112
+
and cache_hash <> ? and cache_hash <> '' and state = 'ready'
113
+
order by created_at desc
114
+
limit 1`, repoDID, engine, key, excludeHash))
115
+
}
116
+
117
+
func (d *DB) TouchCacheEntry(ctx context.Context, id string, now time.Time) error {
118
+
_, err := d.ExecContext(ctx, `
119
+
update cache_entries set last_used_at = ? where id = ? and state = 'ready'`, now.UnixNano(), id)
120
+
return err
121
+
}
122
+
123
+
func (d *DB) ClaimCacheEntry(ctx context.Context, id, expectedState string, expectedLastUsed time.Time) (bool, error) {
124
+
result, err := d.ExecContext(ctx, `
125
+
update cache_entries
126
+
set state = 'deleting'
127
+
where id = ? and state = ? and last_used_at = ?`,
128
+
id, expectedState, expectedLastUsed.UnixNano())
129
+
if err != nil {
130
+
return false, err
131
+
}
132
+
changed, err := result.RowsAffected()
133
+
return changed == 1, err
134
+
}
135
+
136
+
func (d *DB) RestoreCacheEntryState(ctx context.Context, id, state string) error {
137
+
_, err := d.ExecContext(ctx, `
138
+
update cache_entries set state = ? where id = ? and state = 'deleting'`, state, id)
139
+
return err
140
+
}
141
+
142
+
func (d *DB) ExpiredCacheEntries(ctx context.Context, readyBefore, pendingBefore time.Time, limit int) ([]CacheEntry, error) {
143
+
ready, err := d.queryCacheEntries(ctx, `
144
+
select `+cacheEntryColumns+`
145
+
from cache_entries
146
+
where state = 'ready' and last_used_at < ?
147
+
order by last_used_at
148
+
limit ?`, readyBefore.UnixNano(), limit)
149
+
if err != nil {
150
+
return nil, err
151
+
}
152
+
recovery, err := d.queryCacheEntries(ctx, `
153
+
select `+cacheEntryColumns+`
154
+
from cache_entries
155
+
where state in ('pending', 'deleting') and created_at < ?
156
+
order by created_at
157
+
limit ?`, pendingBefore.UnixNano(), limit)
158
+
if err != nil {
159
+
return nil, err
160
+
}
161
+
entries := append(ready, recovery...)
162
+
sort.Slice(entries, func(i, j int) bool {
163
+
left, right := entries[i].CreatedAt, entries[j].CreatedAt
164
+
if entries[i].State == "ready" {
165
+
left = entries[i].LastUsedAt
166
+
}
167
+
if entries[j].State == "ready" {
168
+
right = entries[j].LastUsedAt
169
+
}
170
+
return left.Before(right)
171
+
})
172
+
if len(entries) > limit {
173
+
entries = entries[:limit]
174
+
}
175
+
return entries, nil
176
+
}
177
+
178
+
func (d *DB) queryCacheEntries(ctx context.Context, query string, args ...any) ([]CacheEntry, error) {
179
+
rows, err := d.QueryContext(ctx, query, args...)
180
+
if err != nil {
181
+
return nil, err
182
+
}
183
+
defer rows.Close()
184
+
var entries []CacheEntry
185
+
for rows.Next() {
186
+
entry, err := scanCacheEntry(rows)
187
+
if err != nil {
188
+
return nil, err
189
+
}
190
+
entries = append(entries, *entry)
191
+
}
192
+
return entries, rows.Err()
193
+
}
194
+
195
+
func (d *DB) DeleteCacheEntry(ctx context.Context, id string) error {
196
+
_, err := d.ExecContext(ctx, `delete from cache_entries where id = ?`, id)
197
+
return err
198
+
}
199
+
200
+
func (d *DB) CacheUsageByOwner(ctx context.Context, ownerDID string) (bytes, count int64, err error) {
201
+
err = d.QueryRowContext(ctx, `
202
+
select coalesce(sum(size_bytes), 0), count(*)
203
+
from cache_entries
204
+
where owner_did = ? and state in ('ready', 'deleting')`, ownerDID).Scan(&bytes, &count)
205
+
return bytes, count, err
206
+
}
207
+
208
+
type cacheEntryScanner interface {
209
+
Scan(dest ...any) error
210
+
}
211
+
212
+
func scanCacheEntry(row cacheEntryScanner) (*CacheEntry, error) {
213
+
var entry CacheEntry
214
+
var createdAt, lastUsedAt int64
215
+
if err := row.Scan(
216
+
&entry.ID,
217
+
&entry.StorageKey,
218
+
&entry.OwnerDID,
219
+
&entry.RepoDID,
220
+
&entry.Engine,
221
+
&entry.CacheKey,
222
+
&entry.CacheHash,
223
+
&entry.SizeBytes,
224
+
&entry.State,
225
+
&createdAt,
226
+
&lastUsedAt,
227
+
); err != nil {
228
+
return nil, err
229
+
}
230
+
entry.CreatedAt = time.Unix(0, createdAt)
231
+
entry.LastUsedAt = time.Unix(0, lastUsedAt)
232
+
return &entry, nil
233
+
}
+302
spindle/db/cache_test.go
+302
spindle/db/cache_test.go
···
1
+
package db
2
+
3
+
import (
4
+
"context"
5
+
"database/sql"
6
+
"errors"
7
+
"testing"
8
+
"time"
9
+
)
10
+
11
+
func testCacheEntry(id, hash, state string, createdAt time.Time) CacheEntry {
12
+
return CacheEntry{
13
+
ID: id,
14
+
StorageKey: "objects/" + id,
15
+
OwnerDID: "did:plc:owner",
16
+
RepoDID: "did:plc:repo",
17
+
Engine: "microvm",
18
+
CacheKey: "dependencies",
19
+
CacheHash: hash,
20
+
SizeBytes: 10,
21
+
State: state,
22
+
CreatedAt: createdAt,
23
+
LastUsedAt: createdAt,
24
+
}
25
+
}
26
+
27
+
func insertTestCacheEntry(t *testing.T, d *DB, entry CacheEntry) {
28
+
t.Helper()
29
+
if err := d.InsertCacheEntry(context.Background(), entry); err != nil {
30
+
t.Fatalf("InsertCacheEntry(%s): %v", entry.ID, err)
31
+
}
32
+
}
33
+
34
+
func TestCacheEntryLookup(t *testing.T) {
35
+
ctx := context.Background()
36
+
d := newTestDB(t)
37
+
base := time.Date(2026, 1, 2, 3, 4, 5, 6, time.UTC)
38
+
39
+
insertTestCacheEntry(t, d, testCacheEntry("exact-old", "requested", "ready", base))
40
+
insertTestCacheEntry(t, d, testCacheEntry("exact-new", "requested", "ready", base.Add(time.Second)))
41
+
insertTestCacheEntry(t, d, testCacheEntry("exact-pending", "requested", "pending", base.Add(2*time.Second)))
42
+
insertTestCacheEntry(t, d, testCacheEntry("fallback-old", "old-a", "ready", base.Add(3*time.Second)))
43
+
insertTestCacheEntry(t, d, testCacheEntry("fallback-new", "old-b", "ready", base.Add(4*time.Second)))
44
+
insertTestCacheEntry(t, d, testCacheEntry("fallback-pending", "old-c", "pending", base.Add(5*time.Second)))
45
+
46
+
exact, err := d.FindCacheEntry(ctx, "did:plc:repo", "microvm", "dependencies", "requested")
47
+
if err != nil {
48
+
t.Fatalf("FindCacheEntry: %v", err)
49
+
}
50
+
if exact.ID != "exact-new" {
51
+
t.Fatalf("FindCacheEntry returned %q, want exact-new", exact.ID)
52
+
}
53
+
if !exact.CreatedAt.Equal(base.Add(time.Second)) || !exact.LastUsedAt.Equal(base.Add(time.Second)) {
54
+
t.Fatalf("timestamps = (%v, %v), want %v", exact.CreatedAt, exact.LastUsedAt, base.Add(time.Second))
55
+
}
56
+
57
+
fallback, err := d.FindFallbackCacheEntry(ctx, "did:plc:repo", "microvm", "dependencies", "requested")
58
+
if err != nil {
59
+
t.Fatalf("FindFallbackCacheEntry: %v", err)
60
+
}
61
+
if fallback.ID != "fallback-new" {
62
+
t.Fatalf("FindFallbackCacheEntry returned %q, want fallback-new", fallback.ID)
63
+
}
64
+
65
+
if _, err := d.FindCacheEntry(ctx, "did:plc:repo", "microvm", "missing", "requested"); !errors.Is(err, sql.ErrNoRows) {
66
+
t.Fatalf("missing FindCacheEntry error = %v, want sql.ErrNoRows", err)
67
+
}
68
+
}
69
+
70
+
func TestMarkCacheEntryReadySupersedesMatchingReady(t *testing.T) {
71
+
ctx := context.Background()
72
+
d := newTestDB(t)
73
+
base := time.Date(2026, 1, 3, 4, 5, 6, 7, time.UTC)
74
+
first := testCacheEntry("first-completion", "same-hash", "pending", base)
75
+
second := testCacheEntry("second-completion", "same-hash", "pending", base.Add(time.Second))
76
+
insertTestCacheEntry(t, d, first)
77
+
insertTestCacheEntry(t, d, second)
78
+
79
+
superseded, err := d.MarkCacheEntryReady(ctx, first.ID, 100, base.Add(2*time.Second))
80
+
if err != nil {
81
+
t.Fatalf("mark first ready: %v", err)
82
+
}
83
+
if len(superseded) != 0 {
84
+
t.Fatalf("first completion superseded %d entries, want none", len(superseded))
85
+
}
86
+
superseded, err = d.MarkCacheEntryReady(ctx, second.ID, 200, base.Add(3*time.Second))
87
+
if err != nil {
88
+
t.Fatalf("mark second ready: %v", err)
89
+
}
90
+
if len(superseded) != 1 || superseded[0].ID != first.ID || superseded[0].State != "deleting" {
91
+
t.Fatalf("second completion superseded %+v, want deleting %s", superseded, first.ID)
92
+
}
93
+
94
+
ready, err := d.FindCacheEntry(ctx, second.RepoDID, second.Engine, second.CacheKey, second.CacheHash)
95
+
if err != nil {
96
+
t.Fatalf("find surviving ready entry: %v", err)
97
+
}
98
+
if ready.ID != second.ID || ready.SizeBytes != 200 || !ready.LastUsedAt.Equal(base.Add(3*time.Second)) {
99
+
t.Fatalf("surviving entry = %+v, want %s (200 bytes)", ready, second.ID)
100
+
}
101
+
}
102
+
103
+
func TestMarkCacheEntryReadyConcurrentCompletions(t *testing.T) {
104
+
ctx := context.Background()
105
+
d := newTestDB(t)
106
+
base := time.Date(2026, 1, 4, 5, 6, 7, 8, time.UTC)
107
+
first := testCacheEntry("concurrent-a", "same-hash", "pending", base)
108
+
second := testCacheEntry("concurrent-b", "same-hash", "pending", base.Add(time.Second))
109
+
insertTestCacheEntry(t, d, first)
110
+
insertTestCacheEntry(t, d, second)
111
+
112
+
type result struct {
113
+
superseded []CacheEntry
114
+
err error
115
+
}
116
+
start := make(chan struct{})
117
+
results := make(chan result, 2)
118
+
for _, entry := range []CacheEntry{first, second} {
119
+
entry := entry
120
+
go func() {
121
+
<-start
122
+
superseded, err := d.MarkCacheEntryReady(ctx, entry.ID, 100, base.Add(2*time.Second))
123
+
results <- result{superseded: superseded, err: err}
124
+
}()
125
+
}
126
+
close(start)
127
+
128
+
var superseded []CacheEntry
129
+
for range 2 {
130
+
result := <-results
131
+
if result.err != nil {
132
+
t.Fatalf("concurrent MarkCacheEntryReady: %v", result.err)
133
+
}
134
+
superseded = append(superseded, result.superseded...)
135
+
}
136
+
if len(superseded) != 1 || superseded[0].State != "deleting" {
137
+
t.Fatalf("concurrent completions superseded %+v, want one deleting entry", superseded)
138
+
}
139
+
140
+
var readyCount, deletingCount int
141
+
if err := d.QueryRowContext(ctx, `
142
+
select sum(state = 'ready'), sum(state = 'deleting')
143
+
from cache_entries
144
+
where repo_did = ? and engine = ? and cache_key = ? and cache_hash = ?`,
145
+
first.RepoDID, first.Engine, first.CacheKey, first.CacheHash).Scan(&readyCount, &deletingCount); err != nil {
146
+
t.Fatalf("count completion states: %v", err)
147
+
}
148
+
if readyCount != 1 || deletingCount != 1 {
149
+
t.Fatalf("completion states = %d ready, %d deleting; want one each", readyCount, deletingCount)
150
+
}
151
+
}
152
+
153
+
func TestCacheEntryReadyTouchAndExpiry(t *testing.T) {
154
+
ctx := context.Background()
155
+
d := newTestDB(t)
156
+
base := time.Date(2026, 2, 3, 4, 5, 6, 7, time.UTC)
157
+
158
+
readyOld := testCacheEntry("ready-old", "a", "ready", base)
159
+
readyFresh := testCacheEntry("ready-fresh", "b", "ready", base)
160
+
pendingOld := testCacheEntry("pending-old", "c", "pending", base)
161
+
pendingFresh := testCacheEntry("pending-fresh", "d", "pending", base.Add(20*time.Minute))
162
+
deletingOld := testCacheEntry("deleting-old", "e", "deleting", base)
163
+
deletingFresh := testCacheEntry("deleting-fresh", "f", "deleting", base.Add(20*time.Minute))
164
+
for _, entry := range []CacheEntry{readyOld, readyFresh, pendingOld, pendingFresh, deletingOld, deletingFresh} {
165
+
insertTestCacheEntry(t, d, entry)
166
+
}
167
+
168
+
touchedAt := base.Add(30 * time.Minute)
169
+
if err := d.TouchCacheEntry(ctx, readyFresh.ID, touchedAt); err != nil {
170
+
t.Fatalf("TouchCacheEntry: %v", err)
171
+
}
172
+
173
+
expired, err := d.ExpiredCacheEntries(ctx, base.Add(10*time.Minute), base.Add(10*time.Minute), 10)
174
+
if err != nil {
175
+
t.Fatalf("ExpiredCacheEntries: %v", err)
176
+
}
177
+
got := make(map[string]bool, len(expired))
178
+
for _, entry := range expired {
179
+
got[entry.ID] = true
180
+
}
181
+
if len(got) != 3 || !got[readyOld.ID] || !got[pendingOld.ID] || !got[deletingOld.ID] {
182
+
t.Fatalf("expired IDs = %v, want ready-old, pending-old, and deleting-old", got)
183
+
}
184
+
185
+
limited, err := d.ExpiredCacheEntries(ctx, base.Add(10*time.Minute), base.Add(10*time.Minute), 1)
186
+
if err != nil {
187
+
t.Fatalf("limited ExpiredCacheEntries: %v", err)
188
+
}
189
+
if len(limited) != 1 {
190
+
t.Fatalf("limited expiry count = %d, want 1", len(limited))
191
+
}
192
+
}
193
+
194
+
func TestCacheEntryClaim(t *testing.T) {
195
+
ctx := context.Background()
196
+
d := newTestDB(t)
197
+
base := time.Date(2026, 2, 4, 5, 6, 7, 8, time.UTC)
198
+
entry := testCacheEntry("claim-me", "hash", "ready", base)
199
+
insertTestCacheEntry(t, d, entry)
200
+
201
+
touchedAt := base.Add(time.Minute)
202
+
if err := d.TouchCacheEntry(ctx, entry.ID, touchedAt); err != nil {
203
+
t.Fatalf("TouchCacheEntry: %v", err)
204
+
}
205
+
claimed, err := d.ClaimCacheEntry(ctx, entry.ID, "ready", base)
206
+
if err != nil {
207
+
t.Fatalf("stale ClaimCacheEntry: %v", err)
208
+
}
209
+
if claimed {
210
+
t.Fatal("stale ClaimCacheEntry claimed a touched entry")
211
+
}
212
+
claimed, err = d.ClaimCacheEntry(ctx, entry.ID, "ready", touchedAt)
213
+
if err != nil {
214
+
t.Fatalf("ClaimCacheEntry: %v", err)
215
+
}
216
+
if !claimed {
217
+
t.Fatal("ClaimCacheEntry did not claim unchanged entry")
218
+
}
219
+
220
+
if err := d.TouchCacheEntry(ctx, entry.ID, base.Add(2*time.Minute)); err != nil {
221
+
t.Fatalf("TouchCacheEntry while deleting: %v", err)
222
+
}
223
+
if _, err := d.MarkCacheEntryReady(ctx, entry.ID, 999, base.Add(3*time.Minute)); !errors.Is(err, sql.ErrNoRows) {
224
+
t.Fatalf("MarkCacheEntryReady while deleting error = %v, want sql.ErrNoRows", err)
225
+
}
226
+
var state string
227
+
var lastUsedAt int64
228
+
var sizeBytes int64
229
+
if err := d.QueryRowContext(ctx, `
230
+
select state, last_used_at, size_bytes from cache_entries where id = ?`, entry.ID).Scan(&state, &lastUsedAt, &sizeBytes); err != nil {
231
+
t.Fatalf("query claimed entry: %v", err)
232
+
}
233
+
if state != "deleting" || lastUsedAt != touchedAt.UnixNano() || sizeBytes != entry.SizeBytes {
234
+
t.Fatalf("claimed entry = state %q, last used %d, size %d; want deleting, %d, %d",
235
+
state, lastUsedAt, sizeBytes, touchedAt.UnixNano(), entry.SizeBytes)
236
+
}
237
+
238
+
if err := d.RestoreCacheEntryState(ctx, entry.ID, "ready"); err != nil {
239
+
t.Fatalf("RestoreCacheEntryState: %v", err)
240
+
}
241
+
restored, err := d.FindCacheEntry(ctx, entry.RepoDID, entry.Engine, entry.CacheKey, entry.CacheHash)
242
+
if err != nil {
243
+
t.Fatalf("find restored entry: %v", err)
244
+
}
245
+
if restored.State != "ready" || !restored.LastUsedAt.Equal(touchedAt) {
246
+
t.Fatalf("restored entry = state %q, last used %v", restored.State, restored.LastUsedAt)
247
+
}
248
+
}
249
+
250
+
func TestCacheEntryDelete(t *testing.T) {
251
+
ctx := context.Background()
252
+
d := newTestDB(t)
253
+
entry := testCacheEntry("delete-me", "hash", "ready", time.Now())
254
+
insertTestCacheEntry(t, d, entry)
255
+
256
+
if err := d.DeleteCacheEntry(ctx, entry.ID); err != nil {
257
+
t.Fatalf("DeleteCacheEntry: %v", err)
258
+
}
259
+
if err := d.DeleteCacheEntry(ctx, entry.ID); err != nil {
260
+
t.Fatalf("second DeleteCacheEntry: %v", err)
261
+
}
262
+
if _, err := d.FindCacheEntry(ctx, entry.RepoDID, entry.Engine, entry.CacheKey, entry.CacheHash); !errors.Is(err, sql.ErrNoRows) {
263
+
t.Fatalf("find deleted error = %v, want sql.ErrNoRows", err)
264
+
}
265
+
}
266
+
267
+
func TestCacheUsageByOwner(t *testing.T) {
268
+
ctx := context.Background()
269
+
d := newTestDB(t)
270
+
base := time.Date(2026, 3, 4, 5, 6, 7, 8, time.UTC)
271
+
272
+
first := testCacheEntry("owner-ready-a", "a", "ready", base)
273
+
first.SizeBytes = 40
274
+
second := testCacheEntry("owner-ready-b", "b", "ready", base)
275
+
second.SizeBytes = 2
276
+
deleting := testCacheEntry("owner-deleting", "old", "deleting", base)
277
+
deleting.SizeBytes = 3
278
+
pending := testCacheEntry("owner-pending", "c", "pending", base)
279
+
pending.SizeBytes = 1000
280
+
other := testCacheEntry("other-ready", "d", "ready", base)
281
+
other.OwnerDID = "did:plc:other"
282
+
other.SizeBytes = 500
283
+
for _, entry := range []CacheEntry{first, second, deleting, pending, other} {
284
+
insertTestCacheEntry(t, d, entry)
285
+
}
286
+
287
+
bytes, count, err := d.CacheUsageByOwner(ctx, "did:plc:owner")
288
+
if err != nil {
289
+
t.Fatalf("CacheUsageByOwner: %v", err)
290
+
}
291
+
if bytes != 45 || count != 3 {
292
+
t.Fatalf("usage = (%d bytes, %d entries), want (45, 3)", bytes, count)
293
+
}
294
+
295
+
bytes, count, err = d.CacheUsageByOwner(ctx, "did:plc:missing")
296
+
if err != nil {
297
+
t.Fatalf("empty CacheUsageByOwner: %v", err)
298
+
}
299
+
if bytes != 0 || count != 0 {
300
+
t.Fatalf("empty usage = (%d bytes, %d entries), want zero", bytes, count)
301
+
}
302
+
}
+25
-1
spindle/db/db.go
+25
-1
spindle/db/db.go
···
107
107
updated_at text not null
108
108
);
109
109
110
+
create table if not exists cache_entries (
111
+
id text primary key,
112
+
storage_key text unique not null,
113
+
owner_did text not null,
114
+
repo_did text not null,
115
+
engine text not null,
116
+
cache_key text not null,
117
+
cache_hash text not null,
118
+
size_bytes integer not null default 0,
119
+
state text not null check (state in ('pending', 'ready', 'deleting')),
120
+
created_at integer not null,
121
+
last_used_at integer not null
122
+
);
123
+
124
+
create index if not exists cache_entries_lookup
125
+
on cache_entries (repo_did, engine, cache_key, cache_hash, created_at desc)
126
+
where state = 'ready';
127
+
create index if not exists cache_entries_ready_expiry
128
+
on cache_entries (last_used_at) where state = 'ready';
129
+
create index if not exists cache_entries_pending_expiry
130
+
on cache_entries (created_at) where state in ('pending', 'deleting');
131
+
create index if not exists cache_entries_owner_usage
132
+
on cache_entries (owner_did) where state in ('ready', 'deleting');
133
+
110
134
create table if not exists pipelines (
111
135
id text primary key,
112
136
repo_did text not null,
···
145
169
return nil, err
146
170
}
147
171
148
-
return &DB{db}, nil
172
+
return &DB{DB: db}, nil
149
173
}
150
174
151
175
func runMigrations(_ context.Context, conn *sql.Conn, logger *slog.Logger) error {
+333
spindle/engine/cache.go
+333
spindle/engine/cache.go
···
1
+
package engine
2
+
3
+
import (
4
+
"bufio"
5
+
"bytes"
6
+
"context"
7
+
"crypto/sha256"
8
+
"database/sql"
9
+
"encoding/hex"
10
+
"errors"
11
+
"fmt"
12
+
"io"
13
+
"log/slog"
14
+
"os/exec"
15
+
"strings"
16
+
"sync"
17
+
"time"
18
+
19
+
"github.com/google/uuid"
20
+
"tangled.org/core/spindle/db"
21
+
22
+
"tangled.org/core/spindle/models"
23
+
"tangled.org/core/spindle/storage"
24
+
)
25
+
26
+
// cache log steps live below the setup step (-1)
27
+
const (
28
+
CacheRestoreStepIdx = -2
29
+
CacheSaveStepIdx = -3
30
+
)
31
+
32
+
type cacheStep struct {
33
+
name string
34
+
command string
35
+
}
36
+
37
+
func (s cacheStep) Name() string { return s.name }
38
+
func (s cacheStep) Command() string { return s.command }
39
+
func (s cacheStep) Kind() models.StepKind { return models.StepKindSystem }
40
+
41
+
var (
42
+
CacheRestoreStep models.Step = cacheStep{name: "restore cache", command: "restore cached paths"}
43
+
CacheSaveStep models.Step = cacheStep{name: "save cache", command: "persist changed paths"}
44
+
)
45
+
46
+
type CacheRunner interface {
47
+
RestoreCache(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, store storage.Storage, caches []ResolvedCache, wfLogger models.WorkflowLogger) error
48
+
SaveCache(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, store storage.Storage, caches []ResolvedCache, wfLogger models.WorkflowLogger) error
49
+
}
50
+
51
+
type ResolvedCache struct {
52
+
Paths []string
53
+
Key string
54
+
Hash string
55
+
SaveKey string
56
+
RestoreID string
57
+
RestoreKey string
58
+
RestoreName string
59
+
CompressionLevel int
60
+
When string
61
+
}
62
+
63
+
func (rc ResolvedCache) saveOn(failed bool) bool {
64
+
return rc.When == "always" || !failed
65
+
}
66
+
67
+
const CacheExitNoPaths = 42
68
+
69
+
// avoids storing an empty archive when zstd is missing
70
+
const CacheExitNoCompressor = 43
71
+
72
+
func AbsolutizePaths(paths []string, workspaceRoot string) []string {
73
+
abs := make([]string, 0, len(paths))
74
+
for _, p := range paths {
75
+
if !strings.HasPrefix(p, "/") {
76
+
p = workspaceRoot + "/" + p
77
+
}
78
+
abs = append(abs, p)
79
+
}
80
+
return abs
81
+
}
82
+
83
+
// run tar from / so absolute paths survive extraction
84
+
// tar exits nonzero on missing paths, so only existing ones reach it
85
+
func CacheSaveScript(paths []string, workspaceRoot string, compressionLevel int) string {
86
+
trimmed := make([]string, 0, len(paths))
87
+
for _, p := range AbsolutizePaths(paths, workspaceRoot) {
88
+
trimmed = append(trimmed, strings.TrimPrefix(p, "/"))
89
+
}
90
+
tail := fmt.Sprintf(`tar -cf - -C / "$@" | %s`, CacheCompressCmd(compressionLevel))
91
+
return fmt.Sprintf(`set -o pipefail
92
+
command -v zstd >/dev/null 2>&1 || { echo "zstd not found in image; cannot save cache" >&2; exit %d; }
93
+
set --
94
+
for p in %s; do [ -e "/$p" ] && set -- "$@" "$p"; done
95
+
if [ $# -eq 0 ]; then echo "no cache paths exist; skipping" >&2; exit %d; fi
96
+
%s`, CacheExitNoCompressor, strings.Join(trimmed, " "), CacheExitNoPaths, tail)
97
+
}
98
+
99
+
func CacheCompressCmd(level int) string {
100
+
if level == 0 {
101
+
return "zstd -T0 -5"
102
+
}
103
+
return fmt.Sprintf("zstd -T0 -%d", level)
104
+
}
105
+
106
+
// old entries might still be gzip, so detect them instead of trusting config
107
+
func CacheDecompressCmd(br *bufio.Reader) string {
108
+
head, _ := br.Peek(4)
109
+
if bytes.HasPrefix(head, []byte{0x1f, 0x8b}) {
110
+
return "gzip -dc"
111
+
}
112
+
return "zstd -dc"
113
+
}
114
+
115
+
// Put keeps draining the pipe after it returns, so the guest writer never
116
+
// blocks on a full pipe.
117
+
type CacheUpload struct {
118
+
Writer *io.PipeWriter
119
+
done chan error
120
+
}
121
+
122
+
func NewCacheUpload(ctx context.Context, store storage.Storage, key string) *CacheUpload {
123
+
pr, pw := io.Pipe()
124
+
u := &CacheUpload{Writer: pw, done: make(chan error, 1)}
125
+
go func() {
126
+
err := store.Put(ctx, key, pr)
127
+
_, _ = io.Copy(io.Discard, pr)
128
+
u.done <- err
129
+
}()
130
+
return u
131
+
}
132
+
133
+
func (u *CacheUpload) Abort(err error) {
134
+
u.Writer.CloseWithError(err)
135
+
<-u.done
136
+
}
137
+
138
+
// storage treats EOF as a complete archive, so only a clean exec may Finish
139
+
func (u *CacheUpload) Finish() error {
140
+
u.Writer.Close()
141
+
return <-u.done
142
+
}
143
+
144
+
type indexedCacheStore struct {
145
+
storage.Storage
146
+
index *db.DB
147
+
logger *slog.Logger
148
+
mu sync.Mutex
149
+
entries map[string]string // storage key -> cache entry id
150
+
}
151
+
152
+
func (s *indexedCacheStore) Get(ctx context.Context, key string) (io.ReadCloser, error) {
153
+
s.mu.Lock()
154
+
id, ok := s.entries[key]
155
+
s.mu.Unlock()
156
+
r, err := s.Storage.Get(ctx, key)
157
+
if err != nil {
158
+
if ok && errors.Is(err, storage.ErrNotExist) {
159
+
_ = s.index.DeleteCacheEntry(context.WithoutCancel(ctx), id)
160
+
}
161
+
return nil, err
162
+
}
163
+
if ok {
164
+
if err := s.index.TouchCacheEntry(ctx, id, time.Now()); err != nil {
165
+
s.logger.Warn("cache usage update failed", "id", id, "err", err)
166
+
}
167
+
}
168
+
return r, nil
169
+
}
170
+
171
+
func (s *indexedCacheStore) Put(ctx context.Context, key string, r io.Reader) error {
172
+
s.mu.Lock()
173
+
id, ok := s.entries[key]
174
+
s.mu.Unlock()
175
+
if !ok {
176
+
return fmt.Errorf("cache metadata missing for %q", key)
177
+
}
178
+
counted := &countingReader{r: r}
179
+
if err := s.Storage.Put(ctx, key, counted); err != nil {
180
+
return err
181
+
}
182
+
superseded, err := s.index.MarkCacheEntryReady(ctx, id, counted.n, time.Now())
183
+
if err != nil {
184
+
_ = s.Storage.Delete(context.WithoutCancel(ctx), key)
185
+
return fmt.Errorf("mark cache ready: %w", err)
186
+
}
187
+
// completed saves leave the pending map so cleanup keeps their object
188
+
s.mu.Lock()
189
+
delete(s.entries, key)
190
+
s.mu.Unlock()
191
+
cleanupCtx := context.WithoutCancel(ctx)
192
+
for _, old := range superseded {
193
+
s.deleteEntry(cleanupCtx, old.StorageKey, old.ID, "replaced")
194
+
}
195
+
return nil
196
+
}
197
+
198
+
func (s *indexedCacheStore) cleanup(ctx context.Context) {
199
+
s.mu.Lock()
200
+
defer s.mu.Unlock()
201
+
for key, id := range s.entries {
202
+
s.deleteEntry(ctx, key, id, "incomplete")
203
+
}
204
+
}
205
+
206
+
func (s *indexedCacheStore) deleteEntry(ctx context.Context, key, id, reason string) {
207
+
if err := s.Storage.Delete(ctx, key); err != nil {
208
+
s.logger.Warn("delete "+reason+" cache failed", "key", key, "err", err)
209
+
return
210
+
}
211
+
if err := s.index.DeleteCacheEntry(ctx, id); err != nil {
212
+
s.logger.Warn("delete "+reason+" cache metadata failed", "id", id, "err", err)
213
+
}
214
+
}
215
+
216
+
type countingReader struct {
217
+
r io.Reader
218
+
n int64
219
+
}
220
+
221
+
func (r *countingReader) Read(p []byte) (int, error) {
222
+
n, err := r.r.Read(p)
223
+
r.n += int64(n)
224
+
return n, err
225
+
}
226
+
227
+
func cacheStoreForRestore(base storage.Storage, index *db.DB, l *slog.Logger, entries []ResolvedCache) storage.Storage {
228
+
mapped := make(map[string]string, len(entries))
229
+
for _, entry := range entries {
230
+
if entry.RestoreKey != "" {
231
+
mapped[entry.RestoreKey] = entry.RestoreID
232
+
}
233
+
}
234
+
return &indexedCacheStore{Storage: base, index: index, logger: l, entries: mapped}
235
+
}
236
+
237
+
func prepareCacheSaves(ctx context.Context, base storage.Storage, index *db.DB, l *slog.Logger, ownerDID, repoDID, engineName string, entries []ResolvedCache) (*indexedCacheStore, error) {
238
+
saveStore := &indexedCacheStore{
239
+
Storage: base,
240
+
index: index,
241
+
logger: l,
242
+
entries: make(map[string]string, len(entries)),
243
+
}
244
+
now := time.Now()
245
+
for i := range entries {
246
+
id := uuid.NewString()
247
+
entries[i].SaveKey = "objects/" + id
248
+
if err := index.InsertCacheEntry(ctx, db.CacheEntry{
249
+
ID: id,
250
+
StorageKey: entries[i].SaveKey,
251
+
OwnerDID: ownerDID,
252
+
RepoDID: repoDID,
253
+
Engine: engineName,
254
+
CacheKey: entries[i].Key,
255
+
CacheHash: entries[i].Hash,
256
+
State: "pending",
257
+
CreatedAt: now,
258
+
LastUsedAt: now,
259
+
}); err != nil {
260
+
saveStore.cleanup(context.WithoutCancel(ctx))
261
+
return nil, err
262
+
}
263
+
saveStore.entries[entries[i].SaveKey] = id
264
+
}
265
+
return saveStore, nil
266
+
}
267
+
268
+
// on a hash miss, the newest older generation still warms the build
269
+
// unusable entries degrade to a plain miss
270
+
func ResolveCaches(ctx context.Context, l *slog.Logger, index *db.DB, repoDID, engine, repoPath, rev string, entries []models.CacheEntry) []ResolvedCache {
271
+
resolved := make([]ResolvedCache, 0, len(entries))
272
+
for _, entry := range entries {
273
+
hash := ""
274
+
if len(entry.Hash) > 0 && repoPath != "" {
275
+
sum, missing := hashKeyFiles(ctx, repoPath, rev, entry.Hash)
276
+
for _, m := range missing {
277
+
l.Warn("cache hash file not in repo", "key", entry.Key, "path", m)
278
+
}
279
+
hash = sum
280
+
}
281
+
rc := ResolvedCache{
282
+
Paths: entry.Paths,
283
+
Key: entry.Key,
284
+
Hash: hash,
285
+
CompressionLevel: entry.CompressionLevel,
286
+
When: entry.When,
287
+
}
288
+
289
+
found, err := index.FindCacheEntry(ctx, repoDID, engine, entry.Key, hash)
290
+
if errors.Is(err, sql.ErrNoRows) && hash != "" {
291
+
found, err = index.FindFallbackCacheEntry(ctx, repoDID, engine, entry.Key, hash)
292
+
if err == nil {
293
+
rc.RestoreName = entry.Key + "-" + found.CacheHash
294
+
}
295
+
}
296
+
if err != nil && !errors.Is(err, sql.ErrNoRows) {
297
+
l.Warn("cache lookup failed; entry will save but not restore", "key", entry.Key, "err", err)
298
+
} else if err == nil {
299
+
rc.RestoreID = found.ID
300
+
rc.RestoreKey = found.StorageKey
301
+
}
302
+
resolved = append(resolved, rc)
303
+
}
304
+
return resolved
305
+
}
306
+
307
+
func hashKeyFiles(ctx context.Context, repoPath, rev string, paths []string) (string, []string) {
308
+
h := sha256.New()
309
+
var missing []string
310
+
hashed := 0
311
+
for _, p := range paths {
312
+
blob, err := gitBlobId(ctx, repoPath, rev, p)
313
+
if err != nil {
314
+
missing = append(missing, p)
315
+
continue
316
+
}
317
+
fmt.Fprintf(h, "%s=%s\n", p, blob)
318
+
hashed++
319
+
}
320
+
if hashed == 0 {
321
+
return "", missing
322
+
}
323
+
return hex.EncodeToString(h.Sum(nil))[:12], missing
324
+
}
325
+
326
+
// sparse checkouts might not have the file, but the object database does
327
+
func gitBlobId(ctx context.Context, repoPath, rev, path string) (string, error) {
328
+
out, err := exec.CommandContext(ctx, "git", "-C", repoPath, "rev-parse", rev+":"+path).Output()
329
+
if err != nil {
330
+
return "", err
331
+
}
332
+
return strings.TrimSpace(string(out)), nil
333
+
}
+85
spindle/engine/cache_prune.go
+85
spindle/engine/cache_prune.go
···
1
+
package engine
2
+
3
+
import (
4
+
"context"
5
+
"log/slog"
6
+
"time"
7
+
8
+
"tangled.org/core/spindle/db"
9
+
"tangled.org/core/spindle/storage"
10
+
)
11
+
12
+
const (
13
+
cachePruneBatch = 100
14
+
cachePendingMaxAge = time.Hour
15
+
)
16
+
17
+
func StartCachePruner(ctx context.Context, l *slog.Logger, index *db.DB, store storage.Storage, retention, interval time.Duration) {
18
+
if store == nil || interval <= 0 {
19
+
return
20
+
}
21
+
go func() {
22
+
prune := func() {
23
+
total := 0
24
+
for {
25
+
n, err := PruneCaches(ctx, index, store, time.Now(), retention, cachePendingMaxAge, cachePruneBatch)
26
+
total += n
27
+
if err != nil {
28
+
l.Warn("cache prune failed", "count", total, "err", err)
29
+
return
30
+
}
31
+
if n < cachePruneBatch {
32
+
if total > 0 {
33
+
l.Info("pruned cache entries", "count", total)
34
+
}
35
+
return
36
+
}
37
+
}
38
+
}
39
+
prune()
40
+
ticker := time.NewTicker(interval)
41
+
defer ticker.Stop()
42
+
for {
43
+
select {
44
+
case <-ctx.Done():
45
+
return
46
+
case <-ticker.C:
47
+
prune()
48
+
}
49
+
}
50
+
}()
51
+
}
52
+
53
+
func PruneCaches(ctx context.Context, index *db.DB, store storage.Storage, now time.Time, retention, pendingMaxAge time.Duration, limit int) (int, error) {
54
+
readyBefore := time.Unix(0, 0)
55
+
if retention > 0 {
56
+
readyBefore = now.Add(-retention)
57
+
}
58
+
entries, err := index.ExpiredCacheEntries(ctx, readyBefore, now.Add(-pendingMaxAge), limit)
59
+
if err != nil {
60
+
return 0, err
61
+
}
62
+
pruned := 0
63
+
for _, entry := range entries {
64
+
if entry.State != "deleting" {
65
+
claimed, err := index.ClaimCacheEntry(ctx, entry.ID, entry.State, entry.LastUsedAt)
66
+
if err != nil {
67
+
return pruned, err
68
+
}
69
+
if !claimed {
70
+
continue
71
+
}
72
+
}
73
+
if err := store.Delete(ctx, entry.StorageKey); err != nil {
74
+
if entry.State != "deleting" {
75
+
_ = index.RestoreCacheEntryState(context.WithoutCancel(ctx), entry.ID, entry.State)
76
+
}
77
+
return pruned, err
78
+
}
79
+
if err := index.DeleteCacheEntry(ctx, entry.ID); err != nil {
80
+
return pruned, err
81
+
}
82
+
pruned++
83
+
}
84
+
return pruned, nil
85
+
}
+152
spindle/engine/cache_prune_test.go
+152
spindle/engine/cache_prune_test.go
···
1
+
package engine
2
+
3
+
import (
4
+
"context"
5
+
"errors"
6
+
"slices"
7
+
"testing"
8
+
"time"
9
+
)
10
+
11
+
func TestPruneCachesExpiryPolicies(t *testing.T) {
12
+
ctx := context.Background()
13
+
now := time.Date(2026, 5, 6, 7, 8, 9, 0, time.UTC)
14
+
type spec struct {
15
+
id string
16
+
state string
17
+
age time.Duration
18
+
}
19
+
tests := []struct {
20
+
name string
21
+
retention time.Duration
22
+
pendingMax time.Duration
23
+
entries []spec
24
+
wantPruned int
25
+
wantSurvive []string
26
+
}{
27
+
{
28
+
name: "ready entries expire by last use",
29
+
retention: time.Hour,
30
+
pendingMax: 15 * time.Minute,
31
+
entries: []spec{{"old", "ready", 2 * time.Hour}, {"fresh", "ready", 30 * time.Minute}},
32
+
wantPruned: 1,
33
+
wantSurvive: []string{"fresh"},
34
+
},
35
+
{
36
+
name: "zero retention keeps ready entries",
37
+
retention: 0,
38
+
pendingMax: time.Hour,
39
+
entries: []spec{{"ready", "ready", 24 * time.Hour}, {"pending", "pending", 2 * time.Hour}},
40
+
wantPruned: 1,
41
+
wantSurvive: []string{"ready"},
42
+
},
43
+
{
44
+
name: "pending entries expire by age",
45
+
retention: time.Hour,
46
+
pendingMax: time.Hour,
47
+
entries: []spec{{"old", "pending", 2 * time.Hour}, {"fresh", "pending", 10 * time.Minute}},
48
+
wantPruned: 1,
49
+
wantSurvive: []string{"fresh"},
50
+
},
51
+
}
52
+
for _, tt := range tests {
53
+
t.Run(tt.name, func(t *testing.T) {
54
+
d := newCacheTestDB(t)
55
+
store := &fakeStorage{objects: map[string][]byte{}}
56
+
keys := map[string]string{}
57
+
for _, e := range tt.entries {
58
+
entry := cacheTestEntry(e.id, "did:plc:repo", "microvm", "deps", "hash", e.state, now.Add(-e.age))
59
+
insertCacheTestEntry(t, d, entry)
60
+
store.objects[entry.StorageKey] = []byte(e.id)
61
+
keys[e.id] = entry.StorageKey
62
+
}
63
+
64
+
pruned, err := PruneCaches(ctx, d, store, now, tt.retention, tt.pendingMax, 10)
65
+
if err != nil {
66
+
t.Fatalf("PruneCaches: %v", err)
67
+
}
68
+
if pruned != tt.wantPruned {
69
+
t.Fatalf("pruned %d entries, want %d", pruned, tt.wantPruned)
70
+
}
71
+
for _, e := range tt.entries {
72
+
survives := slices.Contains(tt.wantSurvive, e.id)
73
+
if store.has(keys[e.id]) != survives || cacheTestEntryExists(t, d, e.id) != survives {
74
+
t.Fatalf("entry %q survived = %v, want %v", e.id, !survives, survives)
75
+
}
76
+
}
77
+
})
78
+
}
79
+
}
80
+
81
+
func TestPruneCachesSkipsEntryRefreshedAfterScan(t *testing.T) {
82
+
ctx := context.Background()
83
+
d := newCacheTestDB(t)
84
+
now := time.Date(2026, 5, 6, 7, 8, 9, 0, time.UTC)
85
+
first := cacheTestEntry("first", "did:plc:repo", "microvm", "deps", "first", "ready", now.Add(-3*time.Hour))
86
+
refreshed := cacheTestEntry("refreshed", "did:plc:repo", "microvm", "deps", "refreshed", "ready", now.Add(-2*time.Hour))
87
+
insertCacheTestEntry(t, d, first)
88
+
insertCacheTestEntry(t, d, refreshed)
89
+
store := &fakeStorage{objects: map[string][]byte{
90
+
first.StorageKey: []byte("first"),
91
+
refreshed.StorageKey: []byte("refreshed"),
92
+
}}
93
+
store.onDelete = func(key string) {
94
+
if key != first.StorageKey {
95
+
return
96
+
}
97
+
if err := d.TouchCacheEntry(ctx, refreshed.ID, now); err != nil {
98
+
t.Fatalf("TouchCacheEntry: %v", err)
99
+
}
100
+
}
101
+
102
+
pruned, err := PruneCaches(ctx, d, store, now, time.Hour, time.Hour, 10)
103
+
if err != nil {
104
+
t.Fatalf("PruneCaches: %v", err)
105
+
}
106
+
if pruned != 1 {
107
+
t.Fatalf("pruned %d entries, want 1", pruned)
108
+
}
109
+
if !store.has(refreshed.StorageKey) || !cacheTestEntryExists(t, d, refreshed.ID) {
110
+
t.Fatal("entry refreshed after expiry scan was pruned")
111
+
}
112
+
}
113
+
114
+
func TestPruneCachesDeleteFailureRemainsRetryable(t *testing.T) {
115
+
for _, initialState := range []string{"ready", "deleting"} {
116
+
t.Run(initialState, func(t *testing.T) {
117
+
ctx := context.Background()
118
+
d := newCacheTestDB(t)
119
+
now := time.Date(2026, 5, 6, 7, 8, 9, 0, time.UTC)
120
+
entry := cacheTestEntry("retry", "did:plc:repo", "microvm", "deps", "hash", initialState, now.Add(-2*time.Hour))
121
+
insertCacheTestEntry(t, d, entry)
122
+
deleteErr := errors.New("delete failed")
123
+
store := &fakeStorage{
124
+
objects: map[string][]byte{entry.StorageKey: []byte("archive")},
125
+
deleteErr: deleteErr,
126
+
}
127
+
128
+
pruned, err := PruneCaches(ctx, d, store, now, time.Hour, time.Hour, 10)
129
+
if !errors.Is(err, deleteErr) {
130
+
t.Fatalf("first PruneCaches error = %v, want %v", err, deleteErr)
131
+
}
132
+
if pruned != 0 {
133
+
t.Fatalf("first prune count = %d, want 0", pruned)
134
+
}
135
+
state, _, _ := cacheTestEntryState(t, d, entry.ID)
136
+
if state != initialState {
137
+
t.Fatalf("state after delete failure = %q, want %q", state, initialState)
138
+
}
139
+
if !store.has(entry.StorageKey) {
140
+
t.Fatal("failed delete removed object")
141
+
}
142
+
143
+
pruned, err = PruneCaches(ctx, d, store, now, time.Hour, time.Hour, 10)
144
+
if err != nil {
145
+
t.Fatalf("retry PruneCaches: %v", err)
146
+
}
147
+
if pruned != 1 || store.has(entry.StorageKey) || cacheTestEntryExists(t, d, entry.ID) {
148
+
t.Fatalf("retry result = pruned %d, object %v, metadata %v", pruned, store.has(entry.StorageKey), cacheTestEntryExists(t, d, entry.ID))
149
+
}
150
+
})
151
+
}
152
+
}
+376
spindle/engine/cache_test.go
+376
spindle/engine/cache_test.go
···
1
+
package engine
2
+
3
+
import (
4
+
"bufio"
5
+
"bytes"
6
+
"context"
7
+
"errors"
8
+
"io"
9
+
"log/slog"
10
+
"os"
11
+
"os/exec"
12
+
"path/filepath"
13
+
"slices"
14
+
"strings"
15
+
"testing"
16
+
"time"
17
+
18
+
"tangled.org/core/spindle/db"
19
+
"tangled.org/core/spindle/models"
20
+
"tangled.org/core/spindle/storage"
21
+
)
22
+
23
+
type fakeStorage struct {
24
+
objects map[string][]byte
25
+
putErr error
26
+
deleteErr error // fails once, then clears
27
+
onDelete func(string)
28
+
}
29
+
30
+
func (f *fakeStorage) Get(_ context.Context, key string) (io.ReadCloser, error) {
31
+
data, ok := f.objects[key]
32
+
if !ok {
33
+
return nil, storage.ErrNotExist
34
+
}
35
+
return io.NopCloser(bytes.NewReader(data)), nil
36
+
}
37
+
38
+
func (f *fakeStorage) Put(_ context.Context, key string, r io.Reader) error {
39
+
data, err := io.ReadAll(r)
40
+
if err != nil {
41
+
return err
42
+
}
43
+
f.objects[key] = data
44
+
return f.putErr
45
+
}
46
+
47
+
func (f *fakeStorage) Delete(_ context.Context, key string) error {
48
+
if f.onDelete != nil {
49
+
f.onDelete(key)
50
+
}
51
+
if f.deleteErr != nil {
52
+
err := f.deleteErr
53
+
f.deleteErr = nil
54
+
return err
55
+
}
56
+
delete(f.objects, key)
57
+
return nil
58
+
}
59
+
60
+
func (f *fakeStorage) has(key string) bool {
61
+
_, ok := f.objects[key]
62
+
return ok
63
+
}
64
+
65
+
func gitRepo(t *testing.T, files map[string]string) (string, string) {
66
+
t.Helper()
67
+
if _, err := exec.LookPath("git"); err != nil {
68
+
t.Skip("git not available")
69
+
}
70
+
dir := t.TempDir()
71
+
run := func(args ...string) {
72
+
t.Helper()
73
+
cmd := exec.Command("git", append([]string{"-C", dir, "-c", "user.email=t@t", "-c", "user.name=t", "-c", "init.defaultBranch=main"}, args...)...)
74
+
if out, err := cmd.CombinedOutput(); err != nil {
75
+
t.Fatalf("git %v: %v\n%s", args, err, out)
76
+
}
77
+
}
78
+
run("init")
79
+
for name, content := range files {
80
+
p := filepath.Join(dir, name)
81
+
if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
82
+
t.Fatal(err)
83
+
}
84
+
if err := os.WriteFile(p, []byte(content), 0o644); err != nil {
85
+
t.Fatal(err)
86
+
}
87
+
}
88
+
run("add", ".")
89
+
run("commit", "-m", "init")
90
+
return dir, "HEAD"
91
+
}
92
+
93
+
func newCacheTestDB(t *testing.T) *db.DB {
94
+
t.Helper()
95
+
d, err := db.Make(context.Background(), filepath.Join(t.TempDir(), "spindle.db"))
96
+
if err != nil {
97
+
t.Fatalf("db.Make: %v", err)
98
+
}
99
+
t.Cleanup(func() { _ = d.Close() })
100
+
return d
101
+
}
102
+
103
+
func cacheTestEntry(id, repoDID, engine, key, hash, state string, at time.Time) db.CacheEntry {
104
+
return db.CacheEntry{
105
+
ID: id,
106
+
StorageKey: "objects/" + id,
107
+
OwnerDID: "did:plc:owner",
108
+
RepoDID: repoDID,
109
+
Engine: engine,
110
+
CacheKey: key,
111
+
CacheHash: hash,
112
+
State: state,
113
+
CreatedAt: at,
114
+
LastUsedAt: at,
115
+
}
116
+
}
117
+
118
+
func insertCacheTestEntry(t *testing.T, d *db.DB, entry db.CacheEntry) {
119
+
t.Helper()
120
+
if err := d.InsertCacheEntry(context.Background(), entry); err != nil {
121
+
t.Fatalf("InsertCacheEntry(%s): %v", entry.ID, err)
122
+
}
123
+
}
124
+
125
+
func cacheTestEntryState(t *testing.T, d *db.DB, id string) (string, int64, time.Time) {
126
+
t.Helper()
127
+
var state string
128
+
var size, lastUsed int64
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
+
t.Fatalf("query cache entry %s: %v", id, err)
131
+
}
132
+
return state, size, time.Unix(0, lastUsed)
133
+
}
134
+
135
+
func cacheTestEntryExists(t *testing.T, d *db.DB, id string) bool {
136
+
t.Helper()
137
+
var exists bool
138
+
if err := d.QueryRow(`select exists(select 1 from cache_entries where id = ?)`, id).Scan(&exists); err != nil {
139
+
t.Fatalf("query cache entry existence %s: %v", id, err)
140
+
}
141
+
return exists
142
+
}
143
+
144
+
var discardLogger = slog.New(slog.NewTextHandler(io.Discard, nil))
145
+
146
+
func TestResolveCachesExactUnhashed(t *testing.T) {
147
+
ctx := context.Background()
148
+
d := newCacheTestDB(t)
149
+
base := time.Date(2026, 4, 5, 6, 7, 8, 0, time.UTC)
150
+
151
+
insertCacheTestEntry(t, d, cacheTestEntry("exact", "did:plc:repo", "microvm", "deps", "", "ready", base))
152
+
insertCacheTestEntry(t, d, cacheTestEntry("other-repo", "did:plc:other", "microvm", "deps", "", "ready", base.Add(time.Hour)))
153
+
insertCacheTestEntry(t, d, cacheTestEntry("other-engine", "did:plc:repo", "nixery", "deps", "", "ready", base.Add(time.Hour)))
154
+
insertCacheTestEntry(t, d, cacheTestEntry("pending", "did:plc:repo", "microvm", "deps", "", "pending", base.Add(time.Hour)))
155
+
insertCacheTestEntry(t, d, cacheTestEntry("hashed-only", "did:plc:repo", "microvm", "tools", "old", "ready", base))
156
+
157
+
resolved := ResolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", "", "", []models.CacheEntry{
158
+
{Key: "deps", Paths: []string{"/x"}, CompressionLevel: 7, When: "always"},
159
+
{Key: "tools", Paths: []string{"/y"}},
160
+
})
161
+
if len(resolved) != 2 {
162
+
t.Fatalf("got %d resolved entries, want 2", len(resolved))
163
+
}
164
+
got := resolved[0]
165
+
if got.RestoreID != "exact" || got.RestoreKey != "objects/exact" || got.RestoreName != "" {
166
+
t.Fatalf("exact restore = (%q, %q, %q)", got.RestoreID, got.RestoreKey, got.RestoreName)
167
+
}
168
+
if got.SaveKey != "" {
169
+
t.Fatalf("resolve allocated a save key %q", got.SaveKey)
170
+
}
171
+
if got.CompressionLevel != 7 || got.When != "always" {
172
+
t.Fatalf("resolved metadata = %+v", got)
173
+
}
174
+
if resolved[1].RestoreKey != "" {
175
+
t.Fatalf("unhashed entry used hashed fallback %q", resolved[1].RestoreKey)
176
+
}
177
+
}
178
+
179
+
func TestResolveCachesHashExactAndFallback(t *testing.T) {
180
+
ctx := context.Background()
181
+
d := newCacheTestDB(t)
182
+
repoPath, rev := gitRepo(t, map[string]string{"go.sum": "v1 contents", "go.mod": "module x"})
183
+
request := []models.CacheEntry{{Key: "go-mod", Hash: []string{"go.sum", "go.mod"}, Paths: []string{"/x"}}}
184
+
185
+
first := ResolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", repoPath, rev, request)[0]
186
+
if first.Hash == "" {
187
+
t.Fatal("hash is empty")
188
+
}
189
+
190
+
base := time.Date(2026, 4, 5, 6, 7, 8, 0, time.UTC)
191
+
insertCacheTestEntry(t, d, cacheTestEntry("exact", "did:plc:repo", "microvm", "go-mod", first.Hash, "ready", base.Add(time.Minute)))
192
+
insertCacheTestEntry(t, d, cacheTestEntry("sibling", "did:plc:repo", "microvm", "go-modules", "newer", "ready", base.Add(time.Hour)))
193
+
194
+
exact := ResolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", repoPath, rev, request)[0]
195
+
if exact.RestoreID != "exact" || exact.RestoreKey != "objects/exact" || exact.RestoreName != "" {
196
+
t.Fatalf("exact generation restore = %+v", exact)
197
+
}
198
+
199
+
if err := os.WriteFile(filepath.Join(repoPath, "go.sum"), []byte("v2 contents"), 0o644); err != nil {
200
+
t.Fatal(err)
201
+
}
202
+
commit := exec.Command("git", "-C", repoPath, "-c", "user.email=t@t", "-c", "user.name=t", "commit", "-qam", "bump")
203
+
if out, err := commit.CombinedOutput(); err != nil {
204
+
t.Fatalf("commit: %v\n%s", err, out)
205
+
}
206
+
rotated := ResolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", repoPath, rev, request)[0]
207
+
if rotated.Hash == first.Hash {
208
+
t.Fatal("lockfile change did not rotate the cache hash")
209
+
}
210
+
if rotated.RestoreID != "exact" || rotated.RestoreKey != "objects/exact" || rotated.RestoreName != "go-mod-"+first.Hash {
211
+
t.Fatalf("fallback restore = %+v", rotated)
212
+
}
213
+
}
214
+
215
+
func TestResolveCachesMissingHashFiles(t *testing.T) {
216
+
ctx := context.Background()
217
+
d := newCacheTestDB(t)
218
+
repoPath, rev := gitRepo(t, map[string]string{"go.mod": "module x"})
219
+
220
+
partial := ResolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", repoPath, rev, []models.CacheEntry{
221
+
{Key: "go-mod", Hash: []string{"go.sum", "go.mod"}, Paths: []string{"/x"}},
222
+
})[0]
223
+
if len(partial.Hash) != 12 {
224
+
t.Fatalf("partial hash = %q, want 12 characters", partial.Hash)
225
+
}
226
+
227
+
none := ResolveCaches(ctx, discardLogger, d, "did:plc:repo", "microvm", repoPath, rev, []models.CacheEntry{
228
+
{Key: "go-mod", Hash: []string{"nope.lock"}, Paths: []string{"/x"}},
229
+
})[0]
230
+
if none.Hash != "" || none.RestoreKey != "" {
231
+
t.Fatalf("all-missing result = hash %q, restore %q", none.Hash, none.RestoreKey)
232
+
}
233
+
}
234
+
235
+
func TestIndexedCacheStoreSaveLifecycle(t *testing.T) {
236
+
ctx := context.Background()
237
+
d := newCacheTestDB(t)
238
+
old := cacheTestEntry("superseded", "did:plc:repo", "microvm", "deps", "hash", "ready", time.Date(2026, 1, 2, 3, 4, 5, 0, time.UTC))
239
+
insertCacheTestEntry(t, d, old)
240
+
base := &fakeStorage{objects: map[string][]byte{old.StorageKey: []byte("old")}}
241
+
entries := []ResolvedCache{{Key: "deps", Hash: "hash"}}
242
+
243
+
store, err := prepareCacheSaves(ctx, base, d, discardLogger, "did:plc:owner", "did:plc:repo", "microvm", entries)
244
+
if err != nil {
245
+
t.Fatalf("prepareCacheSaves: %v", err)
246
+
}
247
+
saveKey := entries[0].SaveKey
248
+
id := store.entries[saveKey]
249
+
250
+
payload := []byte("cache archive bytes")
251
+
if err := store.Put(ctx, saveKey, bytes.NewReader(payload)); err != nil {
252
+
t.Fatalf("indexed Put: %v", err)
253
+
}
254
+
state, size, _ := cacheTestEntryState(t, d, id)
255
+
if state != "ready" || size != int64(len(payload)) {
256
+
t.Fatalf("saved metadata = state %q, size %d", state, size)
257
+
}
258
+
if got := base.objects[saveKey]; !bytes.Equal(got, payload) {
259
+
t.Fatalf("stored payload = %q, want %q", got, payload)
260
+
}
261
+
if base.has(old.StorageKey) || cacheTestEntryExists(t, d, old.ID) {
262
+
t.Fatal("completed save left the superseded generation")
263
+
}
264
+
store.cleanup(ctx)
265
+
if !base.has(saveKey) || !cacheTestEntryExists(t, d, id) {
266
+
t.Fatal("cleanup removed a completed cache")
267
+
}
268
+
}
269
+
270
+
func TestIndexedCacheStoreRestoreTouchesEntry(t *testing.T) {
271
+
ctx := context.Background()
272
+
d := newCacheTestDB(t)
273
+
old := time.Date(2025, 1, 2, 3, 4, 5, 0, time.UTC)
274
+
entry := cacheTestEntry("restore", "did:plc:repo", "microvm", "deps", "hash", "ready", old)
275
+
insertCacheTestEntry(t, d, entry)
276
+
base := &fakeStorage{objects: map[string][]byte{entry.StorageKey: []byte("archive")}}
277
+
store := cacheStoreForRestore(base, d, discardLogger, []ResolvedCache{{RestoreID: entry.ID, RestoreKey: entry.StorageKey}})
278
+
279
+
r, err := store.Get(ctx, entry.StorageKey)
280
+
if err != nil {
281
+
t.Fatalf("indexed Get: %v", err)
282
+
}
283
+
data, err := io.ReadAll(r)
284
+
if err != nil {
285
+
t.Fatalf("read restored object: %v", err)
286
+
}
287
+
if err := r.Close(); err != nil {
288
+
t.Fatalf("close restored object: %v", err)
289
+
}
290
+
if string(data) != "archive" {
291
+
t.Fatalf("restored payload = %q", data)
292
+
}
293
+
_, _, touched := cacheTestEntryState(t, d, entry.ID)
294
+
if !touched.After(old) {
295
+
t.Fatalf("last used = %v, want after %v", touched, old)
296
+
}
297
+
}
298
+
299
+
func TestIndexedCacheStoreFailedPutCleanup(t *testing.T) {
300
+
ctx := context.Background()
301
+
d := newCacheTestDB(t)
302
+
putErr := errors.New("put failed")
303
+
base := &fakeStorage{objects: make(map[string][]byte), putErr: putErr}
304
+
entries := []ResolvedCache{{Key: "deps"}}
305
+
store, err := prepareCacheSaves(ctx, base, d, discardLogger, "did:plc:owner", "did:plc:repo", "microvm", entries)
306
+
if err != nil {
307
+
t.Fatalf("prepareCacheSaves: %v", err)
308
+
}
309
+
saveKey := entries[0].SaveKey
310
+
311
+
if err := store.Put(ctx, saveKey, strings.NewReader("partial")); !errors.Is(err, putErr) {
312
+
t.Fatalf("indexed Put error = %v, want %v", err, putErr)
313
+
}
314
+
store.cleanup(ctx)
315
+
if base.has(saveKey) {
316
+
t.Fatal("cleanup left partial object")
317
+
}
318
+
if cacheTestEntryExists(t, d, store.entries[saveKey]) {
319
+
t.Fatal("cleanup left pending metadata")
320
+
}
321
+
}
322
+
323
+
func TestIndexedCacheStoreRejectsUnpreparedKey(t *testing.T) {
324
+
store := &indexedCacheStore{}
325
+
if err := store.Put(context.Background(), "objects/missing", strings.NewReader("data")); err == nil {
326
+
t.Fatal("Put accepted a key without pending metadata")
327
+
}
328
+
}
329
+
330
+
func TestAbsolutizePaths(t *testing.T) {
331
+
got := AbsolutizePaths([]string{"node_modules", "/root/.cache", ".gocache"}, "/workspace/repo")
332
+
want := []string{"/workspace/repo/node_modules", "/root/.cache", "/workspace/repo/.gocache"}
333
+
if !slices.Equal(got, want) {
334
+
t.Fatalf("got %v, want %v", got, want)
335
+
}
336
+
}
337
+
338
+
func TestCacheDecompressCmd(t *testing.T) {
339
+
cases := []struct {
340
+
name string
341
+
head []byte
342
+
want string
343
+
}{
344
+
{"zstd", []byte{0x28, 0xb5, 0x2f, 0xfd, 0x00}, "zstd -dc"},
345
+
{"gzip", []byte{0x1f, 0x8b, 0x08, 0x00}, "gzip -dc"},
346
+
{"empty", nil, "zstd -dc"},
347
+
}
348
+
for _, tc := range cases {
349
+
got := CacheDecompressCmd(bufio.NewReader(bytes.NewReader(tc.head)))
350
+
if got != tc.want {
351
+
t.Errorf("%s: got %q, want %q", tc.name, got, tc.want)
352
+
}
353
+
}
354
+
}
355
+
356
+
func TestResolvedCacheSaveOn(t *testing.T) {
357
+
cases := []struct {
358
+
when string
359
+
onFail, onPass bool
360
+
}{
361
+
{"", false, true},
362
+
{"on-success", false, true},
363
+
{"always", true, true},
364
+
}
365
+
for _, tc := range cases {
366
+
rc := ResolvedCache{When: tc.when}
367
+
if got := rc.saveOn(true); got != tc.onFail {
368
+
t.Errorf("when=%q failed run: got %v, want %v", tc.when, got, tc.onFail)
369
+
}
370
+
if got := rc.saveOn(false); got != tc.onPass {
371
+
t.Errorf("when=%q passing run: got %v, want %v", tc.when, got, tc.onPass)
372
+
}
373
+
}
374
+
}
375
+
376
+
var _ storage.Storage = (*fakeStorage)(nil)
+80
-2
spindle/engine/engine.go
+80
-2
spindle/engine/engine.go
···
13
13
"tangled.org/core/spindle/db"
14
14
"tangled.org/core/spindle/models"
15
15
"tangled.org/core/spindle/secrets"
16
+
"tangled.org/core/spindle/storage"
16
17
)
17
18
18
19
var (
···
66
67
FinalizeWorkflow(ctx context.Context, wid models.WorkflowId, wf *models.Workflow, wfLogger models.WorkflowLogger) error
67
68
}
68
69
69
-
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) {
70
+
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) {
70
71
l.Info("starting all workflows in parallel", "pipeline", pipelineId)
71
72
73
+
isTrustedRepo := pipeline.TrustedSource && pipeline.RepoDid != ""
72
74
var allSecrets []secrets.UnlockedSecret
73
75
// never pass secrets to pipelines that run untrusted (e.g. fork) code
74
-
if pipeline.TrustedSource && pipeline.RepoDid != "" {
76
+
if isTrustedRepo {
75
77
if res, err := vault.GetSecretsUnlocked(ctx, secrets.RepoIdentifier(pipeline.RepoDid.String())); err == nil {
76
78
allSecrets = res
77
79
}
78
80
} else if !pipeline.TrustedSource {
79
81
l.Info("skipping secrets for untrusted pipeline source", "pipeline", pipelineId)
80
82
}
83
+
// untrusted runs cant read or write shared caches
84
+
cacheOwnerDID := ""
85
+
cacheEnabled := cacheStore != nil && isTrustedRepo
86
+
if cacheEnabled {
87
+
repo, err := db.GetRepoByDid(pipeline.RepoDid)
88
+
if err != nil {
89
+
l.Warn("cache owner lookup failed; caching disabled", "repo", pipeline.RepoDid, "err", err)
90
+
cacheEnabled = false
91
+
} else {
92
+
cacheOwnerDID = repo.Owner.String()
93
+
}
94
+
} else if cacheStore != nil && !pipeline.TrustedSource {
95
+
l.Info("skipping caches for untrusted pipeline source", "pipeline", pipelineId)
96
+
}
97
+
98
+
// hash the checked commit, not the working tree
99
+
cacheRepoPath, cacheRev := "", ""
100
+
if tm := pipeline.TriggerMetadata; tm != nil {
101
+
if rev, err := models.ExtractCommitSHA(*tm); err == nil {
102
+
did := pipeline.RepoDid.String()
103
+
if tm.SourceRepo != nil && *tm.SourceRepo != "" {
104
+
did = *tm.SourceRepo
105
+
}
106
+
cacheRepoPath, cacheRev = filepath.Join(cfg.Server.RepoDir, did), rev
107
+
} else {
108
+
l.Warn("cannot resolve pipeline commit; cache hashing disabled", "err", err)
109
+
}
110
+
}
81
111
82
112
secretValues := make([]string, len(allSecrets))
83
113
for i, s := range allSecrets {
···
105
135
var wg sync.WaitGroup
106
136
for eng, wfs := range pipeline.Workflows {
107
137
workflowTimeout := eng.WorkflowTimeout()
138
+
cacheRunner, cachesSupported := eng.(CacheRunner)
108
139
l.Info("using workflow timeout", "timeout", workflowTimeout)
109
140
110
141
for _, w := range wfs {
···
191
222
}
192
223
defer eng.DestroyWorkflow(ctx, wid)
193
224
225
+
ctx, cancel := context.WithTimeout(ctx, workflowTimeout)
226
+
defer cancel()
227
+
228
+
var resolvedCaches []ResolvedCache
229
+
if cacheEnabled && len(w.Caches) > 0 {
230
+
if cachesSupported {
231
+
resolvedCaches = ResolveCaches(ctx, l, db, pipeline.RepoDid.String(), w.Engine, cacheRepoPath, cacheRev, w.Caches)
232
+
wfLogger.ControlWriter(CacheRestoreStepIdx, CacheRestoreStep, models.StepStatusStart).Write([]byte{0})
233
+
// caches are an optimization, never a reason to fail the workflow
234
+
restoreStore := cacheStoreForRestore(cacheStore, db, l, resolvedCaches)
235
+
if err := cacheRunner.RestoreCache(ctx, wid, &w, restoreStore, resolvedCaches, wfLogger); err != nil {
236
+
l.Warn("cache restore failed", "wid", wid, "err", err)
237
+
}
238
+
wfLogger.ControlWriter(CacheRestoreStepIdx, CacheRestoreStep, models.StepStatusEnd).Write([]byte{0})
239
+
} else {
240
+
l.Warn("engine does not support caches, skipping restore", "wid", wid)
241
+
}
242
+
}
243
+
244
+
// dont save on timeouts, their context is already dead
245
+
saveCaches := func(failed bool) {
246
+
toSave := resolvedCaches[:0]
247
+
for _, rc := range resolvedCaches {
248
+
if rc.saveOn(failed) {
249
+
toSave = append(toSave, rc)
250
+
}
251
+
}
252
+
if len(toSave) == 0 {
253
+
return
254
+
}
255
+
wfLogger.ControlWriter(CacheSaveStepIdx, CacheSaveStep, models.StepStatusStart).Write([]byte{0})
256
+
saveStore, err := prepareCacheSaves(ctx, cacheStore, db, l, cacheOwnerDID, pipeline.RepoDid.String(), w.Engine, toSave)
257
+
if err != nil {
258
+
l.Warn("cache metadata setup failed", "wid", wid, "err", err)
259
+
} else {
260
+
if err := cacheRunner.SaveCache(ctx, wid, &w, saveStore, toSave, wfLogger); err != nil {
261
+
l.Warn("cache save failed", "wid", wid, "err", err)
262
+
}
263
+
saveStore.cleanup(context.WithoutCancel(ctx))
264
+
}
265
+
wfLogger.ControlWriter(CacheSaveStepIdx, CacheSaveStep, models.StepStatusEnd).Write([]byte{0})
266
+
}
194
267
for stepIdx, step := range w.Steps {
195
268
if wfLogger != nil {
196
269
wfLogger.
···
207
280
}
208
281
209
282
if err != nil {
283
+
if !errors.Is(err, ErrTimedOut) && !errors.Is(wfCtx.Err(), context.DeadlineExceeded) && !isCanceled(wfCtx) {
284
+
saveCaches(true)
285
+
}
210
286
writeWfError(db, n, l, wfCtx, wid, "running step", err)
211
287
return
212
288
}
213
289
}
290
+
291
+
saveCaches(false)
214
292
215
293
if finalizer, ok := eng.(workflowFinalizer); ok {
216
294
if err := finalizer.FinalizeWorkflow(wfCtx, wid, &w, wfLogger); err != nil {
+3
-3
spindle/engine/engine_test.go
+3
-3
spindle/engine/engine_test.go
···
114
114
}
115
115
116
116
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
117
-
StartWorkflows(logger, nil, cfg, testDB, nil, context.Background(), pipeline, pipelineId)
117
+
StartWorkflows(logger, nil, cfg, testDB, nil, nil, context.Background(), pipeline, pipelineId)
118
118
119
119
eng.mu.Lock()
120
120
setupCalls := append([]models.WorkflowId(nil), eng.setupCalls...)
···
195
195
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
196
196
doneChan := make(chan struct{})
197
197
go func() {
198
-
StartWorkflows(logger, nil, cfg, testDB, nil, context.Background(), pipeline, pipelineId)
198
+
StartWorkflows(logger, nil, cfg, testDB, nil, nil, context.Background(), pipeline, pipelineId)
199
199
close(doneChan)
200
200
}()
201
201
···
250
250
}
251
251
252
252
cfg := &config.Config{Server: config.Server{LogDir: t.TempDir()}}
253
-
StartWorkflows(logger, nil, cfg, testDB, nil, context.Background(), pipeline, pipelineId)
253
+
StartWorkflows(logger, nil, cfg, testDB, nil, nil, context.Background(), pipeline, pipelineId)
254
254
255
255
st, err := testDB.GetStatus(wid)
256
256
if err != nil {
+125
spindle/engines/microvm/cache.go
+125
spindle/engines/microvm/cache.go
···
1
+
package microvm
2
+
3
+
import (
4
+
"bufio"
5
+
"context"
6
+
"fmt"
7
+
"io"
8
+
9
+
agentv1 "tangled.org/core/spindle/agentproto/gen"
10
+
"tangled.org/core/spindle/engine"
11
+
"tangled.org/core/spindle/models"
12
+
"tangled.org/core/spindle/storage"
13
+
)
14
+
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
+
state, ok := wf.Data.(*workflowState)
17
+
if !ok || state == nil || state.Agent == nil {
18
+
return fmt.Errorf("microVM workflow is not connected to agent")
19
+
}
20
+
21
+
out := wfLogger.DataWriter(engine.CacheRestoreStepIdx, "stdout")
22
+
for _, entry := range caches {
23
+
if err := ctx.Err(); err != nil {
24
+
return err
25
+
}
26
+
if entry.RestoreKey == "" {
27
+
fmt.Fprintf(out, "cache %q: miss\n", entry.Key)
28
+
continue
29
+
}
30
+
if entry.RestoreName != "" {
31
+
fmt.Fprintf(out, "cache %q: restoring from %q\n", entry.Key, entry.RestoreName)
32
+
}
33
+
34
+
rc, err := store.Get(ctx, entry.RestoreKey)
35
+
if err != nil {
36
+
fmt.Fprintf(out, "cache %q: fetch failed: %v\n", entry.Key, err)
37
+
continue
38
+
}
39
+
br := bufio.NewReader(rc)
40
+
decompress := engine.CacheDecompressCmd(br)
41
+
42
+
var restored int64
43
+
exit, err := state.Agent.Exec(ctx, AgentExec{
44
+
ID: fmt.Sprintf("%s-cache-restore", wid.String()),
45
+
ExecStart: cacheExecStart(state, fmt.Sprintf("set -o pipefail\n%s | tar -x -C /", decompress)),
46
+
Stdin: &countingReader{r: br, n: &restored},
47
+
Stderr: out,
48
+
})
49
+
rc.Close()
50
+
if err != nil {
51
+
fmt.Fprintf(out, "cache %q: restore failed: %v\n", entry.Key, err)
52
+
continue
53
+
}
54
+
if exit != 0 {
55
+
fmt.Fprintf(out, "cache %q: restore failed: guest exited %d\n", entry.Key, exit)
56
+
continue
57
+
}
58
+
fmt.Fprintf(out, "cache %q: restored %d bytes\n", entry.Key, restored)
59
+
}
60
+
return nil
61
+
}
62
+
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
+
state, ok := wf.Data.(*workflowState)
65
+
if !ok || state == nil || state.Agent == nil {
66
+
return fmt.Errorf("microVM workflow is not connected to agent")
67
+
}
68
+
69
+
out := wfLogger.DataWriter(engine.CacheSaveStepIdx, "stdout")
70
+
for _, entry := range caches {
71
+
if err := ctx.Err(); err != nil {
72
+
return err
73
+
}
74
+
75
+
script := engine.CacheSaveScript(entry.Paths, guestWorkDir, entry.CompressionLevel)
76
+
77
+
up := engine.NewCacheUpload(ctx, store, entry.SaveKey)
78
+
exit, execErr := state.Agent.Exec(ctx, AgentExec{
79
+
ID: fmt.Sprintf("%s-cache-save", wid.String()),
80
+
ExecStart: cacheExecStart(state, script),
81
+
Stdout: up.Writer,
82
+
Stderr: out,
83
+
})
84
+
switch {
85
+
case exit == engine.CacheExitNoPaths:
86
+
up.Abort(fmt.Errorf("guest exited %d", exit))
87
+
fmt.Fprintf(out, "cache %q: nothing to save\n", entry.Key)
88
+
continue
89
+
case exit == engine.CacheExitNoCompressor:
90
+
up.Abort(fmt.Errorf("guest exited %d", exit))
91
+
fmt.Fprintf(out, "cache %q: zstd not available in image; skipping\n", entry.Key)
92
+
continue
93
+
case execErr != nil:
94
+
up.Abort(execErr)
95
+
return fmt.Errorf("save cache %q: %w", entry.Key, execErr)
96
+
case exit != 0:
97
+
up.Abort(fmt.Errorf("guest exited %d", exit))
98
+
return fmt.Errorf("save cache %q: save script exited %d", entry.Key, exit)
99
+
}
100
+
if err := up.Finish(); err != nil {
101
+
return fmt.Errorf("save cache %q: %w", entry.Key, err)
102
+
}
103
+
fmt.Fprintf(out, "cache %q: saved\n", entry.Key)
104
+
}
105
+
return nil
106
+
}
107
+
108
+
func cacheExecStart(state *workflowState, script string) *agentv1.ExecStart {
109
+
return &agentv1.ExecStart{
110
+
Argv: []string{state.ImageSpec.Shell, "-c", script},
111
+
Env: guestBaseEnv(),
112
+
User: guestWorkflowUser,
113
+
}
114
+
}
115
+
116
+
type countingReader struct {
117
+
r io.Reader
118
+
n *int64
119
+
}
120
+
121
+
func (c *countingReader) Read(p []byte) (int, error) {
122
+
n, err := c.r.Read(p)
123
+
*c.n += int64(n)
124
+
return n, err
125
+
}
+18
-6
spindle/engines/microvm/engine.go
+18
-6
spindle/engines/microvm/engine.go
···
43
43
44
44
type cleanupFunc func(context.Context) error
45
45
46
+
// return a fresh slice since callers append to it
47
+
func guestBaseEnv() []string {
48
+
return []string{
49
+
"HOME=/workspace",
50
+
"LOGNAME=" + guestWorkflowUser,
51
+
"PATH=" + guestBasePATH,
52
+
"USER=" + guestWorkflowUser,
53
+
}
54
+
}
55
+
46
56
type Engine struct {
47
57
l *slog.Logger
48
58
cfg *config.Config
···
135
145
}
136
146
swf.Name = twf.Name
137
147
swf.Environment = dwf.Environment
148
+
149
+
for _, entry := range dwf.Cache {
150
+
if err := entry.Validate(); err != nil {
151
+
return nil, err
152
+
}
153
+
}
154
+
swf.Caches = dwf.Cache
138
155
139
156
if tpl.TriggerMetadata != nil {
140
157
if clone := models.BuildCloneStep(twf, *tpl.TriggerMetadata, e.cfg.Server.Dev); clone.Command() != "" {
···
361
378
err := e.activateConfig(execCtx, wid, state, s, wfLogger.DataWriter(idx, "stdout"))
362
379
return e.classifyStepError(ctx, wid, step, state, stderr, vmExited, "Failed to activate config", err)
363
380
}
364
-
env := []string{
365
-
"HOME=/workspace",
366
-
"LOGNAME=" + guestWorkflowUser,
367
-
"PATH=" + guestBasePATH,
368
-
"USER=" + guestWorkflowUser,
369
-
}
381
+
env := guestBaseEnv()
370
382
for k, v := range w.Environment {
371
383
env = append(env, k+"="+v)
372
384
}
+10
-7
spindle/engines/microvm/models.go
+10
-7
spindle/engines/microvm/models.go
···
3
3
import (
4
4
"fmt"
5
5
"slices"
6
+
7
+
"tangled.org/core/spindle/models"
6
8
)
7
9
8
10
type manifestWorkflow struct {
9
-
Image string `yaml:"image"`
10
-
Services map[string]any `yaml:"services"`
11
-
Virtualisation map[string]any `yaml:"virtualisation"`
12
-
Dependencies []string `yaml:"dependencies"`
13
-
Registry map[string]any `yaml:"registry"`
14
-
Environment map[string]string `yaml:"environment"`
15
-
Substituters map[string]string `yaml:"substituters"`
11
+
Image string `yaml:"image"`
12
+
Services map[string]any `yaml:"services"`
13
+
Virtualisation map[string]any `yaml:"virtualisation"`
14
+
Dependencies []string `yaml:"dependencies"`
15
+
Registry map[string]any `yaml:"registry"`
16
+
Environment map[string]string `yaml:"environment"`
17
+
Substituters map[string]string `yaml:"substituters"`
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
+167
spindle/engines/nixery/cache.go
···
1
+
package nixery
2
+
3
+
import (
4
+
"bufio"
5
+
"context"
6
+
"fmt"
7
+
"io"
8
+
9
+
"github.com/docker/docker/api/types"
10
+
"github.com/docker/docker/api/types/container"
11
+
"github.com/docker/docker/pkg/stdcopy"
12
+
13
+
"tangled.org/core/spindle/engine"
14
+
"tangled.org/core/spindle/models"
15
+
"tangled.org/core/spindle/storage"
16
+
)
17
+
18
+
func (e *Engine) baseEnv() EnvVars {
19
+
envs := EnvVars{}
20
+
envs.AddEnv("HOME", homeDir)
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
+
return envs
23
+
}
24
+
25
+
func (e *Engine) execAttached(ctx context.Context, containerID string, opts container.ExecOptions) (string, types.HijackedResponse, error) {
26
+
execResp, err := e.docker.ContainerExecCreate(ctx, containerID, opts)
27
+
if err != nil {
28
+
return "", types.HijackedResponse{}, fmt.Errorf("create exec: %w", err)
29
+
}
30
+
attach, err := e.docker.ContainerExecAttach(ctx, execResp.ID, container.ExecAttachOptions{})
31
+
if err != nil {
32
+
return "", types.HijackedResponse{}, fmt.Errorf("attach exec: %w", err)
33
+
}
34
+
return execResp.ID, attach, nil
35
+
}
36
+
37
+
func (e *Engine) containerID(wf *models.Workflow) (string, error) {
38
+
addl, ok := wf.Data.(addlFields)
39
+
if !ok || addl.container == "" {
40
+
return "", fmt.Errorf("nixery workflow has no container")
41
+
}
42
+
return addl.container, nil
43
+
}
44
+
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
+
containerID, err := e.containerID(wf)
47
+
if err != nil {
48
+
return err
49
+
}
50
+
51
+
out := wfLogger.DataWriter(engine.CacheRestoreStepIdx, "stdout")
52
+
for _, entry := range caches {
53
+
if err := ctx.Err(); err != nil {
54
+
return err
55
+
}
56
+
if entry.RestoreKey == "" {
57
+
fmt.Fprintf(out, "cache %q: miss\n", entry.Key)
58
+
continue
59
+
}
60
+
if entry.RestoreName != "" {
61
+
fmt.Fprintf(out, "cache %q: restoring from %q\n", entry.Key, entry.RestoreName)
62
+
}
63
+
64
+
rc, err := store.Get(ctx, entry.RestoreKey)
65
+
if err != nil {
66
+
fmt.Fprintf(out, "cache %q: fetch failed: %v\n", entry.Key, err)
67
+
continue
68
+
}
69
+
70
+
br := bufio.NewReader(rc)
71
+
execID, attach, err := e.execAttached(ctx, containerID, container.ExecOptions{
72
+
Cmd: []string{"bash", "-c", engine.CacheDecompressCmd(br) + " | tar -x -C /"},
73
+
Env: e.baseEnv(),
74
+
AttachStdin: true,
75
+
AttachStdout: true,
76
+
AttachStderr: true,
77
+
})
78
+
if err != nil {
79
+
rc.Close()
80
+
return fmt.Errorf("restore cache %q: %w", entry.Key, err)
81
+
}
82
+
83
+
// drain this now or tar can block on stderr before reading stdin
84
+
copyDone := make(chan error, 1)
85
+
go func() {
86
+
_, err := io.Copy(attach.Conn, br)
87
+
_ = attach.CloseWrite()
88
+
copyDone <- err
89
+
}()
90
+
_, _ = stdcopy.StdCopy(out, out, attach.Reader)
91
+
copyErr := <-copyDone
92
+
rc.Close()
93
+
attach.Close()
94
+
if copyErr != nil {
95
+
return fmt.Errorf("restore cache %q: stream archive: %w", entry.Key, copyErr)
96
+
}
97
+
98
+
inspect, err := e.docker.ContainerExecInspect(ctx, execID)
99
+
if err != nil {
100
+
return fmt.Errorf("restore cache %q: %w", entry.Key, err)
101
+
}
102
+
if inspect.ExitCode != 0 {
103
+
fmt.Fprintf(out, "cache %q: extract failed (exit %d)\n", entry.Key, inspect.ExitCode)
104
+
continue
105
+
}
106
+
fmt.Fprintf(out, "cache %q: restored\n", entry.Key)
107
+
}
108
+
return nil
109
+
}
110
+
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
+
containerID, err := e.containerID(wf)
113
+
if err != nil {
114
+
return err
115
+
}
116
+
117
+
out := wfLogger.DataWriter(engine.CacheSaveStepIdx, "stdout")
118
+
for _, entry := range caches {
119
+
if err := ctx.Err(); err != nil {
120
+
return err
121
+
}
122
+
123
+
script := engine.CacheSaveScript(entry.Paths, workspaceDir, entry.CompressionLevel)
124
+
125
+
execID, attach, err := e.execAttached(ctx, containerID, container.ExecOptions{
126
+
Cmd: []string{"bash", "-c", script},
127
+
Env: e.baseEnv(),
128
+
AttachStdout: true,
129
+
AttachStderr: true,
130
+
})
131
+
if err != nil {
132
+
return fmt.Errorf("save cache %q: %w", entry.Key, err)
133
+
}
134
+
135
+
up := engine.NewCacheUpload(ctx, store, entry.SaveKey)
136
+
137
+
// StdCopy only returns once the archive is fully written
138
+
_, copyErr := stdcopy.StdCopy(up.Writer, out, attach.Reader)
139
+
attach.Close()
140
+
inspect, inspectErr := e.docker.ContainerExecInspect(ctx, execID)
141
+
142
+
switch {
143
+
case inspectErr == nil && inspect.ExitCode == engine.CacheExitNoPaths:
144
+
up.Abort(fmt.Errorf("no cache paths"))
145
+
fmt.Fprintf(out, "cache %q: nothing to save\n", entry.Key)
146
+
continue
147
+
case inspectErr == nil && inspect.ExitCode == engine.CacheExitNoCompressor:
148
+
up.Abort(fmt.Errorf("zstd not available"))
149
+
fmt.Fprintf(out, "cache %q: zstd not available in image; skipping\n", entry.Key)
150
+
continue
151
+
case copyErr != nil:
152
+
up.Abort(copyErr)
153
+
return fmt.Errorf("save cache %q: stream archive: %w", entry.Key, copyErr)
154
+
case inspectErr != nil:
155
+
up.Abort(inspectErr)
156
+
return fmt.Errorf("save cache %q: %w", entry.Key, inspectErr)
157
+
case inspect.ExitCode != 0:
158
+
up.Abort(fmt.Errorf("exited %d", inspect.ExitCode))
159
+
return fmt.Errorf("save cache %q: tar exited %d", entry.Key, inspect.ExitCode)
160
+
}
161
+
if err := up.Finish(); err != nil {
162
+
return fmt.Errorf("save cache %q: %w", entry.Key, err)
163
+
}
164
+
fmt.Fprintf(out, "cache %q: saved\n", entry.Key)
165
+
}
166
+
return nil
167
+
}
+9
-4
spindle/engines/nixery/engine.go
+9
-4
spindle/engines/nixery/engine.go
···
91
91
} `yaml:"steps"`
92
92
Dependencies map[string][]string `yaml:"dependencies"`
93
93
Environment map[string]string `yaml:"environment"`
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
+
for _, entry := range dwf.Cache {
114
+
if err := entry.Validate(); err != nil {
115
+
return nil, err
116
+
}
117
+
}
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
-
dependencies = path.Join(dependencies, "bash", "git", "coreutils", "nix")
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
-
envs.AddEnv("HOME", homeDir)
413
-
existingPath := "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
414
-
envs.AddEnv("PATH", fmt.Sprintf("%s/.nix-profile/bin:/nix/var/nix/profiles/default/bin:%s", homeDir, existingPath))
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
}
+63
spindle/models/cache.go
+63
spindle/models/cache.go
···
1
+
package models
2
+
3
+
import (
4
+
"fmt"
5
+
"regexp"
6
+
"strings"
7
+
)
8
+
9
+
type CacheEntry struct {
10
+
Key string `yaml:"key"`
11
+
Hash []string `yaml:"hash"`
12
+
Paths []string `yaml:"paths"`
13
+
// 1 is fastest, 19 is smallest, 0 is the zstd default
14
+
CompressionLevel int `yaml:"compression-level"`
15
+
// on-success is the default, always also saves on failed runs
16
+
When string `yaml:"when"`
17
+
}
18
+
19
+
// keys become storage paths, so no slashes
20
+
var cacheKeyRe = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,127}$`)
21
+
22
+
func (c CacheEntry) Validate() error {
23
+
if !cacheKeyRe.MatchString(c.Key) {
24
+
return fmt.Errorf("cache: invalid key %q (allowed: letters, digits, '.', '_', '-')", c.Key)
25
+
}
26
+
for _, f := range c.Hash {
27
+
// rev-parse needs plain repo-relative paths, not pathspecs
28
+
if f == "" || strings.HasPrefix(f, "/") || strings.HasPrefix(f, "..") {
29
+
return fmt.Errorf("cache %q: hash path %q is not repo-relative", c.Key, f)
30
+
}
31
+
if strings.ContainsAny(f, ": \t\n\"'`$\\*?[") || strings.Contains(f, "/../") || strings.HasSuffix(f, "/..") {
32
+
return fmt.Errorf("cache %q: hash path %q contains unsupported characters", c.Key, f)
33
+
}
34
+
}
35
+
if len(c.Paths) == 0 {
36
+
return fmt.Errorf("cache %q: no paths", c.Key)
37
+
}
38
+
if c.CompressionLevel < 0 || c.CompressionLevel > 19 {
39
+
return fmt.Errorf("cache %q: compression-level %d out of range (1-19)", c.Key, c.CompressionLevel)
40
+
}
41
+
switch c.When {
42
+
case "", "on-success", "always":
43
+
default:
44
+
return fmt.Errorf("cache %q: when %q is not one of on-success, always", c.Key, c.When)
45
+
}
46
+
seen := make(map[string]bool, len(c.Paths))
47
+
for _, p := range c.Paths {
48
+
if p == "" {
49
+
return fmt.Errorf("cache %q: empty path", c.Key)
50
+
}
51
+
if strings.ContainsAny(p, " \t\n\"'`$\\") {
52
+
return fmt.Errorf("cache %q: path %q contains unsupported characters", c.Key, p)
53
+
}
54
+
if strings.Contains(p, "..") {
55
+
return fmt.Errorf("cache %q: path %q must not contain '..'", c.Key, p)
56
+
}
57
+
if seen[p] {
58
+
return fmt.Errorf("cache %q: duplicate path %q", c.Key, p)
59
+
}
60
+
seen[p] = true
61
+
}
62
+
return nil
63
+
}
+49
spindle/models/cache_test.go
+49
spindle/models/cache_test.go
···
1
+
package models
2
+
3
+
import (
4
+
"strings"
5
+
"testing"
6
+
)
7
+
8
+
func TestCacheEntryValidate(t *testing.T) {
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
+
if err := valid.Validate(); err != nil {
11
+
t.Fatalf("valid entry: %v", err)
12
+
}
13
+
14
+
cases := []struct {
15
+
name string
16
+
entry CacheEntry
17
+
want string
18
+
}{
19
+
{"empty key", CacheEntry{Key: "", Paths: []string{"/x"}}, "invalid key"},
20
+
{"key with slash", CacheEntry{Key: "a/b", Paths: []string{"/x"}}, "invalid key"},
21
+
{"no paths", CacheEntry{Key: "ok"}, "no paths"},
22
+
{"empty path", CacheEntry{Key: "ok", Paths: []string{""}}, "empty path"},
23
+
{"path with space", CacheEntry{Key: "ok", Paths: []string{"/my dir"}}, "unsupported characters"},
24
+
{"path with quote", CacheEntry{Key: "ok", Paths: []string{"/x'$(rm)"}}, "unsupported characters"},
25
+
{"path traversal", CacheEntry{Key: "ok", Paths: []string{"/x/../y"}}, ".."},
26
+
{"duplicate path", CacheEntry{Key: "ok", Paths: []string{"/x", "/x"}}, "duplicate"},
27
+
{"level too high", CacheEntry{Key: "ok", Paths: []string{"/x"}, CompressionLevel: 20}, "out of range"},
28
+
{"negative level", CacheEntry{Key: "ok", Paths: []string{"/x"}, CompressionLevel: -1}, "out of range"},
29
+
{"bad when", CacheEntry{Key: "ok", Paths: []string{"/x"}, When: "sometimes"}, "not one of"},
30
+
{"absolute hash path", CacheEntry{Key: "ok", Hash: []string{"/go.sum"}, Paths: []string{"/x"}}, "not repo-relative"},
31
+
{"empty hash path", CacheEntry{Key: "ok", Hash: []string{""}, Paths: []string{"/x"}}, "not repo-relative"},
32
+
{"hash path traversal", CacheEntry{Key: "ok", Hash: []string{"../secret"}, Paths: []string{"/x"}}, "not repo-relative"},
33
+
{"hash path inner traversal", CacheEntry{Key: "ok", Hash: []string{"a/../../b"}, Paths: []string{"/x"}}, "unsupported characters"},
34
+
{"hash path with colon", CacheEntry{Key: "ok", Hash: []string{"rev:go.sum"}, Paths: []string{"/x"}}, "unsupported characters"},
35
+
{"hash path with space", CacheEntry{Key: "ok", Hash: []string{"my lock"}, Paths: []string{"/x"}}, "unsupported characters"},
36
+
{"hash path with glob", CacheEntry{Key: "ok", Hash: []string{"*.sum"}, Paths: []string{"/x"}}, "unsupported characters"},
37
+
}
38
+
for _, tc := range cases {
39
+
err := tc.entry.Validate()
40
+
if err == nil || !strings.Contains(err.Error(), tc.want) {
41
+
t.Errorf("%s: got %v, want error containing %q", tc.name, err, tc.want)
42
+
}
43
+
}
44
+
45
+
ok := CacheEntry{Key: "go-mod", Hash: []string{"go.sum"}, Paths: []string{"node_modules", "/root/.cache"}, CompressionLevel: 19, When: "always"}
46
+
if err := ok.Validate(); err != nil {
47
+
t.Errorf("relative and absolute paths should validate, got %v", err)
48
+
}
49
+
}
+2
-3
spindle/models/clone.go
+2
-3
spindle/models/clone.go
···
47
47
return CloneStep{}
48
48
}
49
49
50
-
commitSHA, err := extractCommitSHA(tr)
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
-
// extractCommitSHA extracts the commit SHA from trigger metadata based on trigger type
87
-
func extractCommitSHA(tr tangled.Pipeline_TriggerMetadata) (string, error) {
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
+9
-1
spindle/models/pipeline.go
···
1
1
package models
2
2
3
-
import "github.com/bluesky-social/indigo/atproto/syntax"
3
+
import (
4
+
"github.com/bluesky-social/indigo/atproto/syntax"
5
+
6
+
"tangled.org/core/api/tangled"
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
+
// used to resolve cache hash files against the checkout the workflow builds
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
+
Caches []CacheEntry
39
+
Engine string
32
40
}
+20
-4
spindle/server.go
+20
-4
spindle/server.go
···
41
41
"tangled.org/core/spindle/git"
42
42
"tangled.org/core/spindle/models"
43
43
"tangled.org/core/spindle/secrets"
44
+
"tangled.org/core/spindle/storage"
44
45
"tangled.org/core/spindle/xrpc"
45
46
"tangled.org/core/tid"
46
47
"tangled.org/core/workflow"
···
68
69
res *idresolver.Resolver
69
70
verify repoverify.Verifier
70
71
vault secrets.Manager
72
+
cache storage.Storage
71
73
motd []byte
72
74
motdMu sync.RWMutex
73
75
rootCtx context.Context
···
150
152
151
153
resolver := idresolver.DefaultResolver(cfg.Server.PlcUrl)
152
154
155
+
cacheStore, err := storage.New(ctx, cfg)
156
+
if err != nil {
157
+
return nil, fmt.Errorf("failed to setup cache storage: %w", err)
158
+
}
159
+
if cacheStore != nil {
160
+
logger.Info("cache storage enabled", "backend", cfg.Cache.Backend)
161
+
}
162
+
153
163
spindle := &Spindle{
154
164
jc: jc,
155
165
e: e,
···
161
171
res: resolver,
162
172
verify: repoverify.New(resolver, cfg.Server.Dev),
163
173
vault: vault,
174
+
cache: cacheStore,
164
175
motd: defaultMotd,
165
176
rootCtx: ctx,
166
177
jobWake: make(chan struct{}, 1),
···
222
233
cfg.Server.Tap.AdminPassword = pw
223
234
logger.Info("embedded tap: using random admin password")
224
235
}
236
+
engine.StartCachePruner(ctx, logger, d, cacheStore, cfg.Cache.Retention, cfg.Cache.PruneInterval)
237
+
225
238
spindle.tap = NewTapClient(spindle)
226
239
227
240
return spindle, nil
···
818
831
ewf.Environment = make(map[string]string)
819
832
}
820
833
maps.Copy(ewf.Environment, pipelineEnv)
834
+
835
+
ewf.Engine = w.Engine
821
836
workflows[eng] = append(workflows[eng], *ewf)
822
837
}
823
838
824
-
engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, s.rootCtx, &models.Pipeline{
825
-
RepoDid: syntax.DID(job.RepoDid),
826
-
Workflows: workflows,
827
-
TrustedSource: trustedSource,
839
+
engine.StartWorkflows(log.SubLogger(s.l, "engine"), s.vault, s.cfg, s.db, s.n, s.cache, s.rootCtx, &models.Pipeline{
840
+
RepoDid: syntax.DID(job.RepoDid),
841
+
Workflows: workflows,
842
+
TrustedSource: trustedSource,
843
+
TriggerMetadata: job.Tpl.TriggerMetadata,
828
844
}, pipelineId)
829
845
}
830
846
+87
spindle/storage/disk.go
+87
spindle/storage/disk.go
···
1
+
package storage
2
+
3
+
import (
4
+
"context"
5
+
"fmt"
6
+
"io"
7
+
"os"
8
+
"path/filepath"
9
+
)
10
+
11
+
type Disk struct {
12
+
root string
13
+
}
14
+
15
+
func NewDisk(root string) (*Disk, error) {
16
+
if root == "" {
17
+
return nil, fmt.Errorf("storage: disk backend requires a directory")
18
+
}
19
+
if err := os.MkdirAll(root, 0o755); err != nil {
20
+
return nil, fmt.Errorf("storage: create disk root: %w", err)
21
+
}
22
+
return &Disk{root: filepath.Clean(root)}, nil
23
+
}
24
+
25
+
func (d *Disk) path(key string) (string, error) {
26
+
if err := ValidateKey(key); err != nil {
27
+
return "", err
28
+
}
29
+
return filepath.Join(d.root, filepath.FromSlash(key)), nil
30
+
}
31
+
32
+
func (d *Disk) Get(_ context.Context, key string) (io.ReadCloser, error) {
33
+
p, err := d.path(key)
34
+
if err != nil {
35
+
return nil, err
36
+
}
37
+
f, err := os.Open(p)
38
+
if err != nil {
39
+
if os.IsNotExist(err) {
40
+
return nil, ErrNotExist
41
+
}
42
+
return nil, fmt.Errorf("storage: get %q: %w", key, err)
43
+
}
44
+
return f, nil
45
+
}
46
+
47
+
func (d *Disk) Put(_ context.Context, key string, r io.Reader) error {
48
+
p, err := d.path(key)
49
+
if err != nil {
50
+
return err
51
+
}
52
+
if err := os.MkdirAll(filepath.Dir(p), 0o755); err != nil {
53
+
return fmt.Errorf("storage: put %q: %w", key, err)
54
+
}
55
+
// dont let readers see a partial object
56
+
tmp, err := os.CreateTemp(filepath.Dir(p), ".tmp-*")
57
+
if err != nil {
58
+
return fmt.Errorf("storage: put %q: %w", key, err)
59
+
}
60
+
tmpName := tmp.Name()
61
+
defer os.Remove(tmpName)
62
+
if _, err := io.Copy(tmp, r); err != nil {
63
+
_ = tmp.Close()
64
+
return fmt.Errorf("storage: put %q: %w", key, err)
65
+
}
66
+
if err := tmp.Close(); err != nil {
67
+
return fmt.Errorf("storage: put %q: %w", key, err)
68
+
}
69
+
if err := os.Rename(tmpName, p); err != nil {
70
+
return fmt.Errorf("storage: put %q: %w", key, err)
71
+
}
72
+
return nil
73
+
}
74
+
75
+
func (d *Disk) Delete(_ context.Context, key string) error {
76
+
p, err := d.path(key)
77
+
if err != nil {
78
+
return err
79
+
}
80
+
if err := os.Remove(p); err != nil {
81
+
if os.IsNotExist(err) {
82
+
return nil
83
+
}
84
+
return fmt.Errorf("storage: delete %q: %w", key, err)
85
+
}
86
+
return nil
87
+
}
+102
spindle/storage/s3.go
+102
spindle/storage/s3.go
···
1
+
package storage
2
+
3
+
import (
4
+
"context"
5
+
"errors"
6
+
"fmt"
7
+
"io"
8
+
"strings"
9
+
10
+
awsconfig "github.com/aws/aws-sdk-go-v2/config"
11
+
"github.com/aws/aws-sdk-go-v2/service/s3"
12
+
"github.com/aws/aws-sdk-go-v2/service/s3/types"
13
+
)
14
+
15
+
type S3 struct {
16
+
bucket string
17
+
prefix string
18
+
client *s3.Client
19
+
}
20
+
21
+
func NewS3(ctx context.Context, bucket, prefix string) (*S3, error) {
22
+
if bucket == "" {
23
+
return nil, fmt.Errorf("storage: s3 backend requires a bucket")
24
+
}
25
+
sdkConfig, err := awsconfig.LoadDefaultConfig(ctx)
26
+
if err != nil {
27
+
return nil, fmt.Errorf("storage: load s3 config: %w", err)
28
+
}
29
+
client := s3.NewFromConfig(sdkConfig)
30
+
versioning, err := client.GetBucketVersioning(ctx, &s3.GetBucketVersioningInput{
31
+
Bucket: &bucket,
32
+
})
33
+
if err != nil {
34
+
return nil, fmt.Errorf("storage: check s3 bucket versioning: %w", err)
35
+
}
36
+
if versioning.Status != "" {
37
+
return nil, fmt.Errorf("storage: s3 cache bucket must not use versioning")
38
+
}
39
+
return &S3{
40
+
bucket: bucket,
41
+
prefix: strings.Trim(prefix, "/"),
42
+
client: client,
43
+
}, nil
44
+
}
45
+
46
+
func (s *S3) fullKey(key string) (string, error) {
47
+
if err := ValidateKey(key); err != nil {
48
+
return "", err
49
+
}
50
+
if s.prefix == "" {
51
+
return key, nil
52
+
}
53
+
return s.prefix + "/" + key, nil
54
+
}
55
+
56
+
func (s *S3) Get(ctx context.Context, key string) (io.ReadCloser, error) {
57
+
full, err := s.fullKey(key)
58
+
if err != nil {
59
+
return nil, err
60
+
}
61
+
res, err := s.client.GetObject(ctx, &s3.GetObjectInput{
62
+
Bucket: &s.bucket,
63
+
Key: &full,
64
+
})
65
+
if err != nil {
66
+
var nsk *types.NoSuchKey
67
+
if errors.As(err, &nsk) {
68
+
return nil, ErrNotExist
69
+
}
70
+
return nil, fmt.Errorf("storage: get %q: %w", key, err)
71
+
}
72
+
return res.Body, nil
73
+
}
74
+
75
+
func (s *S3) Put(ctx context.Context, key string, r io.Reader) error {
76
+
full, err := s.fullKey(key)
77
+
if err != nil {
78
+
return err
79
+
}
80
+
if _, err := s.client.PutObject(ctx, &s3.PutObjectInput{
81
+
Bucket: &s.bucket,
82
+
Key: &full,
83
+
Body: r,
84
+
}); err != nil {
85
+
return fmt.Errorf("storage: put %q: %w", key, err)
86
+
}
87
+
return nil
88
+
}
89
+
90
+
func (s *S3) Delete(ctx context.Context, key string) error {
91
+
full, err := s.fullKey(key)
92
+
if err != nil {
93
+
return err
94
+
}
95
+
if _, err := s.client.DeleteObject(ctx, &s3.DeleteObjectInput{
96
+
Bucket: &s.bucket,
97
+
Key: &full,
98
+
}); err != nil {
99
+
return fmt.Errorf("storage: delete %q: %w", key, err)
100
+
}
101
+
return nil
102
+
}
+52
spindle/storage/storage.go
+52
spindle/storage/storage.go
···
1
+
package storage
2
+
3
+
import (
4
+
"context"
5
+
"errors"
6
+
"fmt"
7
+
"io"
8
+
"path/filepath"
9
+
"regexp"
10
+
"strings"
11
+
12
+
"tangled.org/core/spindle/config"
13
+
)
14
+
15
+
var ErrNotExist = errors.New("storage: object does not exist")
16
+
17
+
type Storage interface {
18
+
Get(ctx context.Context, key string) (io.ReadCloser, error)
19
+
Put(ctx context.Context, key string, r io.Reader) error
20
+
Delete(ctx context.Context, key string) error
21
+
}
22
+
23
+
var keyRe = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:/@%+=-]{0,511}$`)
24
+
25
+
func ValidateKey(key string) error {
26
+
if !keyRe.MatchString(key) {
27
+
return fmt.Errorf("storage: invalid key %q", key)
28
+
}
29
+
for _, seg := range strings.Split(key, "/") {
30
+
if seg == "" || seg == "." || seg == ".." {
31
+
return fmt.Errorf("storage: invalid key %q", key)
32
+
}
33
+
}
34
+
return nil
35
+
}
36
+
37
+
func New(ctx context.Context, cfg *config.Config) (Storage, error) {
38
+
switch cfg.Cache.Backend {
39
+
case "":
40
+
return nil, nil
41
+
case "disk":
42
+
dir := cfg.Cache.DiskDir
43
+
if dir == "" {
44
+
dir = filepath.Join(filepath.Dir(cfg.Server.DBPath), "cache")
45
+
}
46
+
return NewDisk(dir)
47
+
case "s3":
48
+
return NewS3(ctx, cfg.Cache.S3Bucket, cfg.Cache.S3Prefix)
49
+
default:
50
+
return nil, fmt.Errorf("storage: unknown backend %q", cfg.Cache.Backend)
51
+
}
52
+
}
+101
spindle/storage/storage_test.go
+101
spindle/storage/storage_test.go
···
1
+
package storage
2
+
3
+
import (
4
+
"context"
5
+
"errors"
6
+
"io"
7
+
"strings"
8
+
"testing"
9
+
)
10
+
11
+
func TestValidateKey(t *testing.T) {
12
+
valid := []string{
13
+
"did:plc:xyz123/go-mod-v1",
14
+
"did:web:spindle.example.com/cache.tar",
15
+
"abc",
16
+
"a/b/c/d",
17
+
}
18
+
for _, k := range valid {
19
+
if err := ValidateKey(k); err != nil {
20
+
t.Errorf("ValidateKey(%q) = %v, want nil", k, err)
21
+
}
22
+
}
23
+
24
+
invalid := []string{
25
+
"",
26
+
"../escape",
27
+
"a/../../b",
28
+
"/leading",
29
+
"trailing/",
30
+
"double//slash",
31
+
"with space",
32
+
"with\\backslash",
33
+
"-leading-dash",
34
+
}
35
+
for _, k := range invalid {
36
+
if err := ValidateKey(k); err == nil {
37
+
t.Errorf("ValidateKey(%q) = nil, want error", k)
38
+
}
39
+
}
40
+
}
41
+
42
+
func TestDiskRoundTrip(t *testing.T) {
43
+
ctx := context.Background()
44
+
d, err := NewDisk(t.TempDir())
45
+
if err != nil {
46
+
t.Fatal(err)
47
+
}
48
+
49
+
key := "did:plc:xyz/go-mod-v1"
50
+
if err := d.Put(ctx, key, strings.NewReader("archive-bytes")); err != nil {
51
+
t.Fatal(err)
52
+
}
53
+
54
+
rc, err := d.Get(ctx, key)
55
+
if err != nil {
56
+
t.Fatal(err)
57
+
}
58
+
got, err := io.ReadAll(rc)
59
+
rc.Close()
60
+
if err != nil {
61
+
t.Fatal(err)
62
+
}
63
+
if string(got) != "archive-bytes" {
64
+
t.Fatalf("got %q, want %q", got, "archive-bytes")
65
+
}
66
+
67
+
// overwrite
68
+
if err := d.Put(ctx, key, strings.NewReader("new-bytes")); err != nil {
69
+
t.Fatal(err)
70
+
}
71
+
rc, _ = d.Get(ctx, key)
72
+
got, _ = io.ReadAll(rc)
73
+
rc.Close()
74
+
if string(got) != "new-bytes" {
75
+
t.Fatalf("overwrite: got %q, want %q", got, "new-bytes")
76
+
}
77
+
78
+
if err := d.Delete(ctx, key); err != nil {
79
+
t.Fatalf("delete: %v", err)
80
+
}
81
+
if _, err := d.Get(ctx, key); !errors.Is(err, ErrNotExist) {
82
+
t.Fatalf("get deleted: got %v, want ErrNotExist", err)
83
+
}
84
+
if err := d.Delete(ctx, key); err != nil {
85
+
t.Fatalf("delete missing: %v", err)
86
+
}
87
+
}
88
+
89
+
func TestDiskRejectsTraversal(t *testing.T) {
90
+
ctx := context.Background()
91
+
d, err := NewDisk(t.TempDir())
92
+
if err != nil {
93
+
t.Fatal(err)
94
+
}
95
+
if err := d.Put(ctx, "../evil", strings.NewReader("x")); err == nil {
96
+
t.Fatal("put with traversal key succeeded")
97
+
}
98
+
if _, err := d.Get(ctx, "../evil"); err == nil {
99
+
t.Fatal("get with traversal key succeeded")
100
+
}
101
+
}