alice.pds.demo.boltless.dev
This repository has no description
1package db
2
3import (
4 "context"
5 "database/sql"
6 "errors"
7 "testing"
8 "time"
9)
10
11func 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
27func 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
34func 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
70func 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
103func 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
153func 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
194func 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
250func 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
267func 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}